Add policies property to cluster resource

Add policies property to cluster resource, policies will be attached
to cluster after cluster creation.

Change-Id: I272fbace441db76339bb6fb689e7f863cca49cad

heat/engine/clients/os/senlin.py

@@ -63,6 +63,10 @@ class SenlinClientPlugin(client_plugin.ClientPlugin):
         cluster = self.client().get_cluster(cluster_name)
         return cluster.id
 
+    def get_policy_id(self, policy_name):
+        policy = self.client().get_policy(policy_name)
+        return policy.id
+
     def is_not_found(self, ex):
         return isinstance(ex, exc.sdkexc.ResourceNotFound)
 
@@ -70,6 +74,26 @@ class SenlinClientPlugin(client_plugin.ClientPlugin):
         return (isinstance(ex, exc.sdkexc.HttpException) and
                 ex.http_status == 400)
 
+    def execute_actions(self, actions):
+        all_executed = True
+        for action in actions:
+            if action['done']:
+                continue
+            all_executed = False
+            if action['action_id'] is None:
+                func = getattr(self.client(), action['func'])
+                ret = func(**action['params'])
+                if isinstance(ret, dict):
+                    action['action_id'] = ret['action']
+                else:
+                    action['action_id'] = ret.location.split('/')[-1]
+            else:
+                ret = self.check_action_status(action['action_id'])
+                action['done'] = ret
+            # Execute these actions one by one.
+            break
+        return all_executed
+
 
 class ProfileConstraint(constraints.BaseCustomConstraint):
     # If name is not unique, will raise exc.sdkexc.HttpException
@@ -87,6 +111,14 @@ class ClusterConstraint(constraints.BaseCustomConstraint):
         client.client(CLIENT_NAME).get_cluster(value)
 
 
+class PolicyConstraint(constraints.BaseCustomConstraint):
+    #  If name is not unique, will raise exc.sdkexc.HttpException
+    expected_exceptions = (exc.sdkexc.HttpException,)
+
+    def validate_with_client(self, client, value):
+        client.client(CLIENT_NAME).get_policy(value)
+
+
 class ProfileTypeConstraint(constraints.BaseCustomConstraint):
 
     expected_exceptions = (exception.StackValidationFailed,)

heat/engine/resources/openstack/senlin/cluster.py

@@ -38,18 +38,24 @@ class Cluster(resource.Resource):
 
     PROPERTIES = (
         NAME, PROFILE, DESIRED_CAPACITY, MIN_SIZE, MAX_SIZE,
-        METADATA, TIMEOUT
+        METADATA, TIMEOUT, POLICIES,
     ) = (
         'name', 'profile', 'desired_capacity', 'min_size', 'max_size',
-        'metadata', 'timeout'
+        'metadata', 'timeout', 'policies',
     )
 
     ATTRIBUTES = (
         ATTR_NAME, ATTR_METADATA, ATTR_NODES, ATTR_DESIRED_CAPACITY,
-        ATTR_MIN_SIZE, ATTR_MAX_SIZE,
+        ATTR_MIN_SIZE, ATTR_MAX_SIZE, ATTR_POLICIES,
     ) = (
         "name", 'metadata', 'nodes', 'desired_capacity',
-        'min_size', 'max_size'
+        'min_size', 'max_size', 'policies',
+    )
+
+    _POLICIES = (
+        P_POLICY, P_ENABLED,
+    ) = (
+        "policy", "enabled",
     )
 
     _CLUSTER_STATUS = (
@@ -115,6 +121,30 @@ class Cluster(resource.Resource):
                 constraints.Range(min=0)
             ]
         ),
+        POLICIES: properties.Schema(
+            properties.Schema.LIST,
+            _('A list of policies to attach to this cluster.'),
+            update_allowed=True,
+            support_status=support.SupportStatus(version='8.0.0'),
+            schema=properties.Schema(
+                properties.Schema.MAP,
+                schema={
+                    P_POLICY: properties.Schema(
+                        properties.Schema.STRING,
+                        _("The name or ID of the policy."),
+                        required=True,
+                        constraints=[
+                            constraints.CustomConstraint('senlin.policy')
+                        ]
+                    ),
+                    P_ENABLED: properties.Schema(
+                        properties.Schema.BOOLEAN,
+                        _("Whether enable this policy on this cluster."),
+                        default=True,
+                    ),
+                }
+            )
+        ),
     }
 
     attributes_schema = {
@@ -142,6 +172,11 @@ class Cluster(resource.Resource):
             _("Max size of the cluster."),
             type=attributes.Schema.INTEGER
         ),
+        ATTR_POLICIES: attributes.Schema(
+            _("Policies attached to the cluster."),
+            type=attributes.Schema.LIST,
+            support_status=support.SupportStatus(version='8.0.0'),
+        ),
     }
 
     def translation_rules(self, props):
@@ -152,10 +187,17 @@ class Cluster(resource.Resource):
                 translation_path=[self.PROFILE],
                 client_plugin=self.client_plugin(),
                 finder='get_profile_id'),
+            translation.TranslationRule(
+                props,
+                translation.TranslationRule.RESOLVE,
+                translation_path=[self.POLICIES, self.P_POLICY],
+                client_plugin=self.client_plugin(),
+                finder='get_policy_id'),
         ]
         return rules
 
     def handle_create(self):
+        actions = []
         params = {
             'name': (self.properties[self.NAME] or
                      self.physical_resource_name()),
@@ -166,13 +208,38 @@ class Cluster(resource.Resource):
             'metadata': self.properties[self.METADATA],
             'timeout': self.properties[self.TIMEOUT]
         }
+        action = {
+            'func': 'create_cluster',
+            'params': params,
+            'action_id': None,
+            'done': False,
+        }
         cluster = self.client().create_cluster(**params)
         action_id = cluster.location.split('/')[-1]
         self.resource_id_set(cluster.id)
-        return action_id
+        action = {
+            'action_id': action_id,
+            'done': False,
+        }
+        actions.append(action)
+        if self.properties[self.POLICIES]:
+            for p in self.properties[self.POLICIES]:
+                params = {
+                    'cluster': cluster.id,
+                    'policy': p[self.P_POLICY],
+                    'enabled': p[self.P_ENABLED],
+                }
+                action = {
+                    'func': 'cluster_attach_policy',
+                    'params': params,
+                    'action_id': None,
+                    'done': False,
+                }
+                actions.append(action)
+        return actions
 
-    def check_create_complete(self, action_id):
+    def check_create_complete(self, actions):
-        return self.client_plugin().check_action_status(action_id)
+        return self.client_plugin().execute_actions(actions)
 
     def handle_delete(self):
         if self.resource_id is not None:
@@ -194,56 +261,94 @@ class Cluster(resource.Resource):
     def handle_update(self, json_snippet, tmpl_diff, prop_diff):
         UPDATE_PROPS = (self.NAME, self.METADATA, self.TIMEOUT, self.PROFILE)
         RESIZE_PROPS = (self.MIN_SIZE, self.MAX_SIZE, self.DESIRED_CAPACITY)
-        updaters = {}
+        actions = []
-        if prop_diff:
+        if not prop_diff:
-            if any(p in prop_diff for p in UPDATE_PROPS):
+            return actions
-                params = dict((k, v) for k, v in six.iteritems(prop_diff)
+        cluster_obj = self.client().get_cluster(self.resource_id)
-                              if k in UPDATE_PROPS)
+        # Update Policies
-                if self.PROFILE in prop_diff:
+        if self.POLICIES in prop_diff:
-                    params.pop(self.PROFILE)
+            old_policies = self.properties[self.POLICIES]
-                    params['profile_id'] = prop_diff[self.PROFILE]
+            new_policies = prop_diff[self.POLICIES]
-                    updaters['cluster_update'] = {
+            old_policy_ids = [p[self.P_POLICY] for p in old_policies]
-                        'params': params,
+            update_policies = [p for p in new_policies
-                        'start': False,
+                               if p[self.P_POLICY] in old_policy_ids]
-                    }
+            update_policy_ids = [p[self.P_POLICY] for p in update_policies]
-            if any(p in prop_diff for p in RESIZE_PROPS):
+            add_policies = [p for p in new_policies
-                params = dict((k, v) for k, v in six.iteritems(prop_diff)
+                            if p[self.P_POLICY] not in old_policy_ids]
-                              if k in RESIZE_PROPS)
+            remove_policies = [p for p in old_policies
-                if self.DESIRED_CAPACITY in prop_diff:
+                               if p[self.P_POLICY] not in update_policy_ids]
-                    params.pop(self.DESIRED_CAPACITY)
+            for p in update_policies:
-                    params['adjustment_type'] = 'EXACT_CAPACITY'
+                params = {
-                    params['number'] = prop_diff.pop(self.DESIRED_CAPACITY)
+                    'policy': p[self.P_POLICY],
-                updaters['cluster_resize'] = {
+                    'cluster': self.resource_id,
-                    'params': params,
+                    'enabled': p[self.P_ENABLED]
-                    'start': False,
                 }
-            return updaters
+                action = {
+                    'func': 'cluster_update_policy',
+                    'params': params,
+                    'action_id': None,
+                    'done': False,
+                }
+                actions.append(action)
+            for p in remove_policies:
+                params = {
+                    'policy': p[self.P_POLICY],
+                    'cluster': self.resource_id,
+                    'enabled': p[self.P_ENABLED]
+                }
+                action = {
+                    'func': 'cluster_detach_policy',
+                    'params': params,
+                    'action_id': None,
+                    'done': False,
+                }
+                actions.append(action)
+            for p in add_policies:
+                params = {
+                    'policy': p[self.P_POLICY],
+                    'cluster': self.resource_id,
+                    'enabled': p[self.P_ENABLED]
+                }
+                action = {
+                    'func': 'cluster_attach_policy',
+                    'params': params,
+                    'action_id': None,
+                    'done': False,
+                }
+                actions.append(action)
+        # Update cluster
+        if any(p in prop_diff for p in UPDATE_PROPS):
+            params = dict((k, v) for k, v in six.iteritems(prop_diff)
+                          if k in UPDATE_PROPS)
+            params['cluster'] = cluster_obj
+            if self.PROFILE in params:
+                params['profile_id'] = params.pop(self.PROFILE)
+            action = {
+                'func': 'update_cluster',
+                'params': params,
+                'action_id': None,
+                'done': False,
+            }
+            actions.append(action)
+        # Resize Cluster
+        if any(p in prop_diff for p in RESIZE_PROPS):
+            params = dict((k, v) for k, v in six.iteritems(prop_diff)
+                          if k in RESIZE_PROPS)
+            if self.DESIRED_CAPACITY in params:
+                params['adjustment_type'] = 'EXACT_CAPACITY'
+                params['number'] = params.pop(self.DESIRED_CAPACITY)
+            params['cluster'] = self.resource_id
+            action = {
+                'func': 'cluster_resize',
+                'params': params,
+                'action_id': None,
+                'done': False,
+            }
+            actions.append(action)
+        return actions
 
-    def check_update_complete(self, updaters):
+    def check_update_complete(self, actions):
-        def start_action(action, params):
+        return self.client_plugin().execute_actions(actions)
-            if action == 'cluster_resize':
-                resp = self.client().cluster_resize(self.resource_id,
-                                                    **params)
-                return resp['action']
-            elif action == 'cluster_update':
-                cluster_obj = self.client().get_cluster(self.resource_id)
-                resp = self.client().update_cluster(cluster_obj,
-                                                    **params)
-                return resp.location.split('/')[-1]
-
-        if not updaters:
-            return True
-        for k, updater in list(updaters.items()):
-            if not updater['start']:
-                action_id = start_action(k, updater['params'])
-                updater['action'] = action_id
-                updater['start'] = True
-            else:
-                ret = self.client_plugin().check_action_status(
-                    updater['action'])
-                if ret:
-                    del updaters[k]
-            return False
 
     def validate(self):
         min_size = self.properties[self.MIN_SIZE]
@@ -271,11 +376,16 @@ class Cluster(resource.Resource):
         if self.resource_id is None:
             return
         cluster = self.client().get_cluster(self.resource_id)
+        if name == self.ATTR_POLICIES:
+            return self.client().cluster_policies(self.resource_id)
         return getattr(cluster, name, None)
 
     def _show_resource(self):
         cluster = self.client().get_cluster(self.resource_id)
-        return cluster.to_dict()
+        cluster_dict = cluster.to_dict()
+        cluster_dict[self.ATTR_POLICIES] = self.client().cluster_policies(
+            self.resource_id)
+        return cluster_dict
 
     def parse_live_resource_data(self, resource_properties, resource_data):
         reality = {}
@@ -283,6 +393,14 @@ class Cluster(resource.Resource):
         for key in self._update_allowed_properties:
             if key == self.PROFILE:
                 value = resource_data.get('profile_id')
+            elif key == self.POLICIES:
+                value = []
+                for p in resource_data.get(self.POLICIES):
+                    v = {
+                        'policy': p.get('policy_id'),
+                        'enabled': p.get('enabled'),
+                    }
+                    value.append(v)
             else:
                 value = resource_data.get(key)
             reality.update({key: value})

heat/engine/resources/openstack/senlin/node.py

@@ -198,25 +198,7 @@ class Node(resource.Resource):
         return actions
 
     def check_update_complete(self, actions):
-        update_complete = True
+        return self.client_plugin().execute_actions(actions)
-        for action in actions:
-            if action['done']:
-                continue
-            update_complete = False
-            if action['action_id'] is None:
-                func = getattr(self.client(), action['func'])
-                ret = func(**action['params'])
-                if isinstance(ret, dict):
-                    action['action_id'] = ret['action']
-                else:
-                    action['action_id'] = ret.location.split('/')[-1]
-            else:
-                ret = self.client_plugin().check_action_status(
-                    action['action_id'])
-                action['done'] = ret
-            # Execute these actions one by one.
-            break
-        return update_complete
 
     def _resolve_attribute(self, name):
         if self.resource_id is None:

heat/tests/clients/test_senlin_client.py

@@ -60,6 +60,14 @@ class SenlinClientPluginTest(common.HeatTestCase):
         self.assertEqual('fake_cluster_id', ret)
         mock_get.assert_called_once_with('fake_cluster')
 
+    def test_get_policy_id(self):
+        mock_policy = mock.Mock(id='fake_policy_id')
+        mock_get = self.patchobject(self.client, 'get_policy',
+                                    return_value=mock_policy)
+        ret = self.plugin.get_policy_id('fake_policy')
+        self.assertEqual('fake_policy_id', ret)
+        mock_get.assert_called_once_with('fake_policy')
+
 
 class ProfileConstraintTest(common.HeatTestCase):
 
@@ -109,6 +117,30 @@ class ClusterConstraintTest(common.HeatTestCase):
         self.assertFalse(self.constraint.validate("CLUSTER_ID", self.ctx))
 
 
+class PolicyConstraintTest(common.HeatTestCase):
+
+    def setUp(self):
+        super(PolicyConstraintTest, self).setUp()
+        self.senlin_client = mock.MagicMock()
+        self.ctx = utils.dummy_context()
+        self.mock_get_policy = mock.Mock()
+        self.ctx.clients.client(
+            'senlin').get_policy = self.mock_get_policy
+        self.constraint = senlin_plugin.PolicyConstraint()
+
+    def test_validate_true(self):
+        self.mock_get_policy.return_value = None
+        self.assertTrue(self.constraint.validate("POLICY_ID", self.ctx))
+
+    def test_validate_false(self):
+        self.mock_get_policy.side_effect = exc.sdkexc.ResourceNotFound(
+            'POLICY_ID')
+        self.assertFalse(self.constraint.validate("POLICY_ID", self.ctx))
+        self.mock_get_policy.side_effect = exc.sdkexc.HttpException(
+            'POLICY_ID')
+        self.assertFalse(self.constraint.validate("POLICY_ID", self.ctx))
+
+
 class ProfileTypeConstraintTest(common.HeatTestCase):
 
     def setUp(self):

heat/tests/openstack/senlin/test_cluster.py

@@ -38,6 +38,9 @@ resources:
     properties:
       name: SenlinCluster
       profile: fake_profile
+      policies:
+        - policy: fake_policy
+          enabled: true
       min_size: 0
       max_size: -1
       desired_capacity: 1
@@ -93,6 +96,8 @@ class SenlinClusterTest(common.HeatTestCase):
                          return_value=self.senlin_mock)
         self.patchobject(senlin.ProfileConstraint, 'validate',
                          return_value=True)
+        self.patchobject(senlin.PolicyConstraint, 'validate',
+                         return_value=True)
         self.fake_cl = FakeCluster()
         self.t = template_format.parse(cluster_stack_template)
 
@@ -107,6 +112,12 @@ class SenlinClusterTest(common.HeatTestCase):
         self.senlin_mock.get_cluster.return_value = self.fake_cl
         self.senlin_mock.get_action.return_value = mock.Mock(
             status='SUCCEEDED')
+        self.senlin_mock.get_policy.return_value = mock.Mock(
+            id='fake_policy_id'
+        )
+        self.senlin_mock.cluster_policies.return_value = [
+            {'policy_id': 'fake_policy_id', 'enabled': True}
+        ]
         scheduler.TaskRunner(cluster.create)()
         self.assertEqual((cluster.CREATE, cluster.COMPLETE),
                          cluster.state)
@@ -115,7 +126,7 @@ class SenlinClusterTest(common.HeatTestCase):
 
     def test_cluster_create_success(self):
         self._create_cluster(self.t)
-        expect_kwargs = {
+        create_cluster_kwargs = {
             'name': 'SenlinCluster',
             'profile_id': 'fake_profile_id',
             'desired_capacity': 1,
@@ -124,9 +135,15 @@ class SenlinClusterTest(common.HeatTestCase):
             'metadata': {'foo': 'bar'},
             'timeout': 3600,
         }
+        attach_policy_kwargs = {
+            'cluster': self.fake_cl.id,
+            'policy': 'fake_policy_id',
+            'enabled': True
+        }
         self.senlin_mock.create_cluster.assert_called_once_with(
-            **expect_kwargs)
+            **create_cluster_kwargs)
-        self.senlin_mock.get_action.assert_called_once_with('fake-action')
+        self.senlin_mock.cluster_attach_policy.assert_called_once_with(
+            **attach_policy_kwargs)
 
     def test_cluster_create_error(self):
         cfg.CONF.set_override('action_retry_limit', 0, enforce_type=True)
@@ -135,6 +152,9 @@ class SenlinClusterTest(common.HeatTestCase):
         mock_action = mock.MagicMock()
         mock_action.status = 'FAILED'
         mock_action.status_reason = 'oops'
+        self.senlin_mock.get_policy.return_value = mock.Mock(
+            id='fake_policy_id'
+        )
         self.senlin_mock.get_action.return_value = mock_action
         create_task = scheduler.TaskRunner(cluster.create)
         ex = self.assertRaises(exception.ResourceFailure, create_task)
@@ -184,8 +204,8 @@ class SenlinClusterTest(common.HeatTestCase):
             'name': 'new_name'
         }
         self.senlin_mock.update_cluster.assert_called_once_with(
-            self.fake_cl, **cluster_update_kwargs)
+            cluster=self.fake_cl, **cluster_update_kwargs)
-        self.assertEqual(2, self.senlin_mock.get_action.call_count)
+        self.assertEqual(3, self.senlin_mock.get_action.call_count)
 
     def test_cluster_update_desire_capacity(self):
         cluster = self._create_cluster(self.t)
@@ -205,8 +225,64 @@ class SenlinClusterTest(common.HeatTestCase):
             'number': 10
         }
         self.senlin_mock.cluster_resize.assert_called_once_with(
-            cluster.resource_id, **cluster_resize_kwargs)
+            cluster=cluster.resource_id, **cluster_resize_kwargs)
-        self.assertEqual(2, self.senlin_mock.get_action.call_count)
+        self.assertEqual(3, self.senlin_mock.get_action.call_count)
+
+    def test_cluster_update_policy_add_remove(self):
+        cluster = self._create_cluster(self.t)
+        # Mock translate rules
+        self.senlin_mock.get_policy.side_effect = [
+            mock.Mock(id='new_policy_id'),
+            mock.Mock(id='fake_policy_id'),
+            mock.Mock(id='new_policy_id'),
+        ]
+        new_t = copy.deepcopy(self.t)
+        props = new_t['resources']['senlin-cluster']['properties']
+        props['policies'] = [{'policy': 'new_policy'}]
+        rsrc_defns = template.Template(new_t).resource_definitions(self.stack)
+        new_cluster = rsrc_defns['senlin-cluster']
+        self.senlin_mock.cluster_detach_policy.return_value = {
+            'action': 'fake-action'}
+        self.senlin_mock.cluster_attach_policy.return_value = {
+            'action': 'fake-action'}
+        self.senlin_mock.get_action.return_value = mock.Mock(
+            status='SUCCEEDED')
+        scheduler.TaskRunner(cluster.update, new_cluster)()
+        self.assertEqual((cluster.UPDATE, cluster.COMPLETE), cluster.state)
+        detach_policy_kwargs = {
+            'policy': 'fake_policy_id',
+            'cluster': cluster.resource_id,
+            'enabled': True,
+        }
+        self.assertEqual(2,
+                         self.senlin_mock.cluster_attach_policy.call_count)
+        self.senlin_mock.cluster_detach_policy.assert_called_once_with(
+            **detach_policy_kwargs)
+        self.assertEqual(0, self.senlin_mock.cluster_update_policy.call_count)
+        self.assertEqual(4, self.senlin_mock.get_action.call_count)
+
+    def test_cluster_update_policy_exists(self):
+        cluster = self._create_cluster(self.t)
+        new_t = copy.deepcopy(self.t)
+        props = new_t['resources']['senlin-cluster']['properties']
+        props['policies'] = [{'policy': 'fake_policy', 'enabled': False}]
+        rsrc_defns = template.Template(new_t).resource_definitions(self.stack)
+        new_cluster = rsrc_defns['senlin-cluster']
+        self.senlin_mock.cluster_update_policy.return_value = {
+            'action': 'fake-action'}
+        self.senlin_mock.get_action.return_value = mock.Mock(
+            status='SUCCEEDED')
+        scheduler.TaskRunner(cluster.update, new_cluster)()
+        self.assertEqual((cluster.UPDATE, cluster.COMPLETE), cluster.state)
+        update_policy_kwargs = {
+            'policy': 'fake_policy_id',
+            'cluster': cluster.resource_id,
+            'enabled': False,
+        }
+        self.senlin_mock.cluster_update_policy.assert_called_once_with(
+            **update_policy_kwargs)
+        self.assertEqual(1, self.senlin_mock.cluster_attach_policy.call_count)
+        self.assertEqual(0, self.senlin_mock.cluster_detach_policy.call_count)
 
     def test_cluster_update_failed(self):
         cluster = self._create_cluster(self.t)
@@ -240,6 +316,7 @@ class SenlinClusterTest(common.HeatTestCase):
             'nodes': ['node1'],
             'profile_name': 'fake_profile',
             'profile_id': 'fake_profile_id',
+            'policies': [{'policy_id': 'fake_policy_id', 'enabled': True}]
         }
         cluster = self._create_cluster(self.t)
         self.assertEqual(self.fake_cl.desired_capacity,
@@ -258,6 +335,7 @@ class SenlinClusterTest(common.HeatTestCase):
             'max_size': -1,
             'min_size': 0,
             'profile': 'fake_profile_id',
+            'policies': [{'policy': 'fake_policy_id', 'enabled': True}]
         }
         cluster = self._create_cluster(self.t)
         self.senlin_mock.get_cluster.return_value = self.fake_cl

setup.cfg

@@ -141,6 +141,7 @@ heat.constraints =
     sahara.image = heat.engine.clients.os.sahara:ImageConstraint
     sahara.plugin = heat.engine.clients.os.sahara:PluginConstraint
     senlin.cluster = heat.engine.clients.os.senlin:ClusterConstraint
+    senlin.policy = heat.engine.clients.os.senlin:PolicyConstraint
     senlin.policy_type = heat.engine.clients.os.senlin:PolicyTypeConstraint
     senlin.profile = heat.engine.clients.os.senlin:ProfileConstraint
     senlin.profile_type = heat.engine.clients.os.senlin:ProfileTypeConstraint