Merge 8f3ac4a into 6e5b453

otter/convergence/service.py

@@ -131,7 +131,7 @@
 from otter.log.intents import err, msg, msg_with_time, with_log
 from otter.models.intents import (
     DeleteGroup, GetScalingGroupInfo, UpdateGroupErrorReasons,
-    UpdateGroupStatus, UpdateServersCache)
+    UpdateGroupStatus, UpdateScalingGroupStatus, UpdateServersCache)
 from otter.models.interface import NoSuchScalingGroupError, ScalingGroupStatus
 from otter.util.timestamp import datetime_to_epoch
 from otter.util.zk import CreateOrSet, DeleteNode, GetChildren, GetStat
@@ -294,8 +294,17 @@ def execute_convergence(tenant_id, group_id, build_timeout, waiting,
     :raise: :obj:`NoSuchScalingGroupError` if the group doesn't exist.
     """
     clean_waiting = _clean_waiting(waiting, group_id)
-    # Gather data
+
+    # Begin convergence by updating group status to ACTIVE
     yield msg("begin-convergence")
+    try:
+        yield Effect(UpdateScalingGroupStatus(tenant_id, group_id,
+                                              ScalingGroupStatus.ACTIVE))
+    except NoSuchScalingGroupError:
+        # Expected for DELETING group. Ignore.
+        pass
+
+    # Gather data
     now_dt = yield Effect(Func(datetime.utcnow))
     all_data = yield msg_with_time(
         "gather-convergence-data",

otter/models/intents.py

@@ -77,6 +77,25 @@ def perform_update_group_status(dispatcher, ugs_intent):
     return ugs_intent.scaling_group.update_status(ugs_intent.status)
 
 
+@attr.s
+class UpdateScalingGroupStatus(object):
+    """
+    Intent to update scaling group status. This is different from
+    :obj:`UpdateGroupStatus` by taking tenant_id and group_id instead
+    of group object
+    """
+    tenant_id = attr.ib()
+    group_id = attr.ib()
+    status = attr.ib()
+
+
+@deferred_performer
+def perform_update_scaling_group_status(log, store, dispatcher, intent):
+    """Perform an :obj:`UpdateScalingGroupStatus`."""
+    group = store.get_scaling_group(log, intent.tenant_id, intent.group_id)
+    return group.update_status(intent.status)
+
+
 @attr.s
 class UpdateServersCache(object):
     """
@@ -135,6 +154,8 @@ def get_model_dispatcher(log, store):
             partial(perform_get_scaling_group_info, log, store),
         DeleteGroup: partial(perform_delete_group, log, store),
         UpdateGroupStatus: perform_update_group_status,
+        UpdateScalingGroupStatus:
+            partial(perform_update_scaling_group_status, log, store),
         UpdateServersCache: perform_update_servers_cache,
         UpdateGroupErrorReasons: perform_update_error_reasons,
         ModifyGroupStatePaused: perform_modify_group_state_paused,

otter/test/convergence/test_service.py

@@ -57,6 +57,7 @@
     GetScalingGroupInfo,
     UpdateGroupErrorReasons,
     UpdateGroupStatus,
+    UpdateScalingGroupStatus,
     UpdateServersCache)
 from otter.models.interface import (
     GroupState, NoSuchScalingGroupError, ScalingGroupStatus)
@@ -857,7 +858,7 @@ def setUp(self):
         self.now = datetime(1970, 1, 1)
         self.waiting = Reference(pmap())
 
-    def get_seq(self, with_cache=True):
+    def get_seq(self, with_cache=True, upd_status_nogroup_err=False):
         exec_seq = [
             (self.gsgi, lambda i: self.gsgi_result),
             (("gacd", self.tenant_id, self.group_id, self.now),
@@ -869,8 +870,16 @@ def get_seq(self, with_cache=True):
                     self.tenant_id, self.group_id, self.now, self.cache),
                  noop)
             )
+        if upd_status_nogroup_err:
+            def handler(i):
+                raise NoSuchScalingGroupError(self.tenant_id, self.group_id)
+        else:
+            handler = noop
         return [
             (Log("begin-convergence", {}), noop),
+            (UpdateScalingGroupStatus(
+                self.tenant_id, self.group_id, ScalingGroupStatus.ACTIVE),
+             handler),
             (Func(datetime.utcnow), lambda i: self.now),
             (MsgWithTime("gather-convergence-data", mock.ANY),
              nested_sequence(exec_seq))
@@ -1068,7 +1077,7 @@ def _plan(dsg, *a, **kwargs):
                                          group_id=self.group_id), noop))
         self.assertEqual(
             # skipping cache update intents returned in get_seq()
-            perform_sequence(self.get_seq(False) + sequence,
+            perform_sequence(self.get_seq(False, True) + sequence,
                              self._invoke(_plan)),
             exec_result)
         # desired capacity was changed to 0

otter/test/models/test_intents.py

@@ -11,8 +11,8 @@
 from otter.log.intents import get_log_dispatcher
 from otter.models.intents import (
     DeleteGroup, GetScalingGroupInfo, ModifyGroupStatePaused,
-    UpdateGroupErrorReasons, UpdateGroupStatus, UpdateServersCache,
-    get_model_dispatcher)
+    UpdateGroupErrorReasons, UpdateGroupStatus, UpdateScalingGroupStatus,
+    UpdateServersCache, get_model_dispatcher)
 from otter.models.interface import (
     GroupState, IScalingGroupCollection, ScalingGroupStatus)
 from otter.test.utils import (
@@ -126,6 +126,20 @@ def test_update_group_status(self):
         self.group.update_status.assert_called_once_with(
             ScalingGroupStatus.ERROR)
 
+    def test_update_scaling_group_status(self):
+        """
+        Performing :obj:`UpdateScalingGroupStatus` calls update_status
+        on group created from tenant_id and group_id in the object
+        """
+        eff = Effect(
+            UpdateScalingGroupStatus("t", "g", ScalingGroupStatus.ERROR))
+        self.group.update_status.return_value = None
+        result = self.perform_with_group(
+            eff, (self.log, 't', 'g'), self.group)
+        self.assertIsNone(result)
+        self.group.update_status.assert_called_once_with(
+            ScalingGroupStatus.ERROR)
+
     @mock.patch('otter.models.intents.CassScalingGroupServersCache',
                 new=EffectServersCache)
     def test_perform_update_servers_cache(self):