Add support for module-reapply command

Server side support for the new 'reapply' command.
This reapplies a given module to all instances that it had
previously been applied to.

Originally, a module designated live-update would automatically
be re-applied whenever it was updated.  Adding a specific
command however, allows operators/users more control over
how the new payload would be distributed.  Old 'modules'
could be left if desired, or updated with the new command.

Scenario tests were updated to test the new command.

DocImpact: update documentation to reflect module-reapply command

Change-Id: I4aea674ebe873a96ed22b5714263d0eea532a4ca
Depends-On: Ic4cc9e9085cb40f1afbec05caeb04886137027a4
Closes-Bug: #1554903

etc/trove/policy.json

@@ -92,5 +92,6 @@
     "module:index": "rule:admin_or_owner",
     "module:show": "rule:admin_or_owner",
     "module:instances": "rule:admin_or_owner",
-    "module:update": "rule:admin_or_owner"
+    "module:update": "rule:admin_or_owner",
+    "module:reapply": "rule:admin_or_owner"
 }

releasenotes/notes/module_reapply-342c0965a4318d4e.yaml

@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    Support for the new 'reapply' command. This allows
+    a given module to be reapplied to all instances that
+    it had previously been applied to. Bug 1554903

trove/common/api.py

@@ -227,6 +227,10 @@ class API(wsgi.Router):
                        controller=modules_resource,
                        action="instances",
                        conditions={'method': ['GET']})
+        mapper.connect("/{tenant_id}/modules/{id}/instances",
+                       controller=modules_resource,
+                       action="reapply",
+                       conditions={'method': ['PUT']})
 
     def _configurations_router(self, mapper):
         parameters_resource = ParametersController().create_resource()

trove/common/cfg.py

@@ -437,6 +437,12 @@ common_opts = [
     cfg.ListOpt('module_types', default=['ping', 'new_relic_license'],
                 help='A list of module types supported. A module type '
                      'corresponds to the name of a ModuleDriver.'),
+    cfg.IntOpt('module_reapply_max_batch_size', default=50,
+               help='The maximum number of instances to reapply a module to '
+                    'at the same time.'),
+    cfg.IntOpt('module_reapply_min_batch_delay', default=2,
+               help='The minimum delay (in seconds) between subsequent '
+                    'module batch reapply executions.'),
     cfg.StrOpt('guest_log_container_name',
                default='database_logs',
                help='Name of container that stores guest log components.'),

trove/common/policy.py

@@ -206,6 +206,8 @@ instance_rules = [
         'module:instances', 'rule:admin_or_owner'),
     policy.RuleDefault(
         'module:update', 'rule:admin_or_owner'),
+    policy.RuleDefault(
+        'module:reapply', 'rule:admin_or_owner'),
 ]
 
 

trove/module/models.py

@@ -30,6 +30,7 @@ from trove.common.i18n import _
 from trove.common import utils
 from trove.datastore import models as datastore_models
 from trove.db import models
+from trove.taskmanager import api as task_api
 
 
 CONF = cfg.CONF
@@ -344,6 +345,12 @@ class Module(object):
         module.updated = datetime.utcnow()
         DBModule.save(module)
 
+    @staticmethod
+    def reapply(context, id, md5, include_clustered,
+                batch_size, batch_delay, force):
+        task_api.API(context).reapply_module(
+            id, md5, include_clustered, batch_size, batch_delay, force)
+
 
 class InstanceModules(object):
 

trove/module/service.py

@@ -19,6 +19,7 @@ import copy
 from oslo_log import log as logging
 
 import trove.common.apischema as apischema
+from trove.common import cfg
 from trove.common import exception
 from trove.common.i18n import _
 from trove.common import pagination
@@ -31,6 +32,7 @@ from trove.module import models
 from trove.module import views
 
 
+CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
 
 
@@ -40,8 +42,8 @@ class ModuleController(wsgi.Controller):
 
     @classmethod
     def authorize_module_action(cls, context, module_rule_name, module):
-        """If a modules in not owned by any particular tenant just check
-        the current tenant is allowed to perform the action.
+        """If a module is not owned by any particular tenant just check
+        that the current tenant is allowed to perform the action.
         """
         if module.tenant_id is not None:
             policy.authorize_on_target(context, 'module:%s' % module_rule_name,
@@ -202,3 +204,30 @@ class ModuleController(wsgi.Controller):
             result_list = pagination.SimplePaginatedDataView(
                 req.url, 'instances', view, marker).data()
         return wsgi.Result(result_list, 200)
+
+    def reapply(self, req, body, tenant_id, id):
+        LOG.info(_("Reapplying module %s to all instances.") % id)
+
+        context = req.environ[wsgi.CONTEXT_KEY]
+        md5 = None
+        if 'md5' in body['reapply']:
+            md5 = body['reapply']['md5']
+        include_clustered = None
+        if 'include_clustered' in body['reapply']:
+            include_clustered = body['reapply']['include_clustered']
+        if 'batch_size' in body['reapply']:
+            batch_size = body['reapply']['batch_size']
+        else:
+            batch_size = CONF.module_reapply_max_batch_size
+        if 'batch_delay' in body['reapply']:
+            batch_delay = body['reapply']['batch_delay']
+        else:
+            batch_delay = CONF.module_reapply_min_batch_delay
+        force = None
+        if 'force' in body['reapply']:
+            force = body['reapply']['force']
+        module = models.Module.load(context, id)
+        self.authorize_module_action(context, 'reapply', module)
+        models.Module.reapply(context, id, md5, include_clustered,
+                              batch_size, batch_delay, force)
+        return wsgi.Result(None, 202)

trove/taskmanager/api.py

@@ -29,6 +29,7 @@ from trove.common.strategies.cluster import strategy
 from trove.guestagent import models as agent_models
 from trove import rpc
 
+
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
 
@@ -268,6 +269,17 @@ class API(object):
         cctxt.cast(self.context, "upgrade_cluster", cluster_id=cluster_id,
                    datastore_version_id=datastore_version_id)
 
+    def reapply_module(self, module_id, md5, include_clustered,
+                       batch_size, batch_delay, force):
+        LOG.debug("Making async call to reapply module %s" % module_id)
+        version = self.API_BASE_VERSION
+
+        cctxt = self.client.prepare(version=version)
+        cctxt.cast(self.context, "reapply_module",
+                   module_id=module_id, md5=md5,
+                   include_clustered=include_clustered,
+                   batch_size=batch_size, batch_delay=batch_delay, force=force)
+
 
 def load(context, manager=None):
     if manager:

trove/taskmanager/manager.py

@@ -416,6 +416,12 @@ class Manager(periodic_task.PeriodicTasks):
             cluster_tasks = models.load_cluster_tasks(context, cluster_id)
             cluster_tasks.delete_cluster(context, cluster_id)
 
+    def reapply_module(self, context, module_id, md5, include_clustered,
+                       batch_size, batch_delay, force):
+        models.ModuleTasks.reapply_module(
+            context, module_id, md5, include_clustered,
+            batch_size, batch_delay, force)
+
     if CONF.exists_notification_transformer:
         @periodic_task.periodic_task
         def publish_exists_event(self, context):

trove/taskmanager/models.py

@@ -58,6 +58,7 @@ from trove.common.notification import (
 import trove.common.remote as remote
 from trove.common.remote import create_cinder_client
 from trove.common.remote import create_dns_client
+from trove.common.remote import create_guest_client
 from trove.common.remote import create_heat_client
 from trove.common import server_group as srv_grp
 from trove.common.strategies.cluster import strategy
@@ -73,9 +74,12 @@ from trove.instance import models as inst_models
 from trove.instance.models import BuiltInstance
 from trove.instance.models import DBInstance
 from trove.instance.models import FreshInstance
+from trove.instance.models import Instance
 from trove.instance.models import InstanceServiceStatus
 from trove.instance.models import InstanceStatus
 from trove.instance.tasks import InstanceTasks
+from trove.module import models as module_models
+from trove.module import views as module_views
 from trove.quota.quota import run_with_quotas
 from trove import rpc
 
@@ -1623,6 +1627,74 @@ class BackupTasks(object):
         LOG.info(_("Deleted backup %s successfully.") % backup_id)
 
 
+class ModuleTasks(object):
+
+    @classmethod
+    def reapply_module(cls, context, module_id, md5, include_clustered,
+                       batch_size, batch_delay, force):
+        """Reapply module."""
+        LOG.info(_("Reapplying module %s.") % module_id)
+
+        batch_size = batch_size or CONF.module_reapply_max_batch_size
+        batch_delay = batch_delay or CONF.module_reapply_min_batch_delay
+        # Don't let non-admin bypass the safeguards
+        if not context.is_admin:
+            batch_size = min(batch_size, CONF.module_reapply_max_batch_size)
+            batch_delay = max(batch_delay, CONF.module_reapply_min_batch_delay)
+        modules = module_models.Modules.load_by_ids(context, [module_id])
+        current_md5 = modules[0].md5
+        LOG.debug("MD5: %s  Force: %s." % (md5, force))
+
+        # Process all the instances
+        instance_modules = module_models.InstanceModules.load_all(
+            context, module_id=module_id, md5=md5)
+        total_count = instance_modules.count()
+        reapply_count = 0
+        skipped_count = 0
+        if instance_modules:
+            module_list = module_views.convert_modules_to_list(modules)
+            for instance_module in instance_modules:
+                instance_id = instance_module.instance_id
+                if (instance_module.md5 != current_md5 or force) and (
+                        not md5 or md5 == instance_module.md5):
+                    instance = BuiltInstanceTasks.load(context, instance_id,
+                                                       needs_server=False)
+                    if instance and (
+                            include_clustered or not instance.cluster_id):
+                        try:
+                            module_models.Modules.validate(
+                                modules, instance.datastore.id,
+                                instance.datastore_version.id)
+                            client = create_guest_client(context, instance_id)
+                            client.module_apply(module_list)
+                            Instance.add_instance_modules(
+                                context, instance_id, modules)
+                            reapply_count += 1
+                        except exception.ModuleInvalid as ex:
+                            LOG.info(_("Skipping: %s") % ex)
+                            skipped_count += 1
+
+                        # Sleep if we've fired off too many in a row.
+                        if (batch_size and
+                                not reapply_count % batch_size and
+                                (reapply_count + skipped_count) < total_count):
+                            LOG.debug("Applied module to %d of %d instances - "
+                                      "sleeping for %ds" % (reapply_count,
+                                                            total_count,
+                                                            batch_delay))
+                            time.sleep(batch_delay)
+                    else:
+                        LOG.debug("Instance '%s' not found or doesn't match "
+                                  "criteria, skipping reapply." % instance_id)
+                        skipped_count += 1
+                else:
+                    LOG.debug("Instance '%s' does not match "
+                              "criteria, skipping reapply." % instance_id)
+                    skipped_count += 1
+        LOG.info(_("Reapplied module to %(num)d instances (skipped %(skip)d).")
+                 % {'num': reapply_count, 'skip': skipped_count})
+
+
 class ResizeVolumeAction(object):
     """Performs volume resize action."""
 

trove/tests/scenario/groups/module_group.py

@@ -375,17 +375,33 @@ class ModuleInstCreateGroup(TestGroup):
         """Check that module-apply works."""
         self.test_runner.run_module_apply()
 
-    @test(runs_after=[module_query_empty])
+    @test(runs_after=[module_apply])
     def module_apply_wrong_module(self):
         """Ensure that module-apply for wrong module fails."""
         self.test_runner.run_module_apply_wrong_module()
 
-    @test(depends_on=[module_apply])
+    @test(depends_on=[module_apply_wrong_module])
+    def module_update_not_live(self):
+        """Ensure updating a non live_update module fails."""
+        self.test_runner.run_module_update_not_live()
+
+    @test(depends_on=[module_apply],
+          runs_after=[module_update_not_live])
     def module_list_instance_after_apply(self):
-        """Check that the instance has one module associated."""
+        """Check that the instance has the modules associated."""
         self.test_runner.run_module_list_instance_after_apply()
 
     @test(runs_after=[module_list_instance_after_apply])
+    def module_apply_live_update(self):
+        """Check that module-apply works for live_update."""
+        self.test_runner.run_module_apply_live_update()
+
+    @test(depends_on=[module_apply_live_update])
+    def module_list_instance_after_apply_live(self):
+        """Check that the instance has the right modules."""
+        self.test_runner.run_module_list_instance_after_apply_live()
+
+    @test(runs_after=[module_list_instance_after_apply_live])
     def module_instances_after_apply(self):
         """Check that the instance shows up in the list."""
         self.test_runner.run_module_instances_after_apply()
@@ -401,13 +417,18 @@ class ModuleInstCreateGroup(TestGroup):
         self.test_runner.run_module_query_after_apply()
 
     @test(runs_after=[module_query_after_apply])
+    def module_update_live_update(self):
+        """Check that update module works on 'live' applied module."""
+        self.test_runner.run_module_update_live_update()
+
+    @test(runs_after=[module_update_live_update])
     def module_apply_another(self):
         """Check that module-apply works for another module."""
         self.test_runner.run_module_apply_another()
 
     @test(depends_on=[module_apply_another])
     def module_list_instance_after_apply_another(self):
-        """Check that the instance has one module associated."""
+        """Check that the instance has the right modules again."""
         self.test_runner.run_module_list_instance_after_apply_another()
 
     @test(runs_after=[module_list_instance_after_apply_another])
@@ -420,7 +441,8 @@ class ModuleInstCreateGroup(TestGroup):
         """Check that the instance count is right after another apply."""
         self.test_runner.run_module_instance_count_after_apply_another()
 
-    @test(depends_on=[module_apply_another])
+    @test(depends_on=[module_apply_another],
+          runs_after=[module_instance_count_after_apply_another])
     def module_query_after_apply_another(self):
         """Check that module-query works after another apply."""
         self.test_runner.run_module_query_after_apply_another()
@@ -431,26 +453,26 @@ class ModuleInstCreateGroup(TestGroup):
         """Check that creating an instance with modules works."""
         self.test_runner.run_create_inst_with_mods()
 
-    @test(runs_after=[module_query_empty])
+    @test(runs_after=[create_inst_with_mods])
     def create_inst_with_wrong_module(self):
         """Ensure that creating an inst with wrong ds mod fails."""
         self.test_runner.run_create_inst_with_wrong_module()
 
-    @test(depends_on=[module_apply])
+    @test(depends_on=[module_apply],
+          runs_after=[create_inst_with_wrong_module])
     def module_delete_applied(self):
         """Ensure that deleting an applied module fails."""
         self.test_runner.run_module_delete_applied()
 
     @test(depends_on=[module_apply],
-          runs_after=[module_list_instance_after_apply,
-                      module_query_after_apply])
+          runs_after=[module_delete_applied])
     def module_remove(self):
         """Check that module-remove works."""
         self.test_runner.run_module_remove()
 
     @test(depends_on=[module_remove])
     def module_query_after_remove(self):
-        """Check that the instance has one module applied after remove."""
+        """Check that the instance has modules applied after remove."""
         self.test_runner.run_module_query_after_remove()
 
     @test(depends_on=[module_remove],
@@ -468,7 +490,7 @@ class ModuleInstCreateGroup(TestGroup):
     @test(depends_on=[module_apply],
           runs_after=[module_apply_another_again])
     def module_query_after_apply_another2(self):
-        """Check that module-query works after second apply."""
+        """Check that module-query works still."""
         self.test_runner.run_module_query_after_apply_another()
 
     @test(depends_on=[module_apply_another_again],
@@ -479,7 +501,7 @@ class ModuleInstCreateGroup(TestGroup):
 
     @test(depends_on=[module_remove_again])
     def module_query_empty_after_again(self):
-        """Check that the inst has one mod applied after 2nd remove."""
+        """Check that the inst has right mod applied after 2nd remove."""
         self.test_runner.run_module_query_after_remove()
 
     @test(depends_on=[module_remove_again],
@@ -533,6 +555,86 @@ class ModuleInstCreateWaitGroup(TestGroup):
         """Ensure that module-delete on auto-applied module fails."""
         self.test_runner.run_module_delete_auto_applied()
 
+    @test(runs_after=[module_delete_auto_applied])
+    def module_list_instance_after_mod_inst(self):
+        """Check that the new instance has the right modules."""
+        self.test_runner.run_module_list_instance_after_mod_inst()
+
+    @test(runs_after=[module_list_instance_after_mod_inst])
+    def module_instances_after_mod_inst(self):
+        """Check that the new instance shows up in the list."""
+        self.test_runner.run_module_instances_after_mod_inst()
+
+    @test(runs_after=[module_instances_after_mod_inst])
+    def module_instance_count_after_mod_inst(self):
+        """Check that the new instance count is right."""
+        self.test_runner.run_module_instance_count_after_mod_inst()
+
+    @test(runs_after=[module_instance_count_after_mod_inst])
+    def module_reapply_with_md5(self):
+        """Check that module reapply with md5 works."""
+        self.test_runner.run_module_reapply_with_md5()
+
+    @test(runs_after=[module_reapply_with_md5])
+    def module_reapply_with_md5_verify(self):
+        """Verify the dates after md5 reapply (no-op)."""
+        self.test_runner.run_module_reapply_with_md5_verify()
+
+    @test(runs_after=[module_reapply_with_md5_verify])
+    def module_list_instance_after_reapply_md5(self):
+        """Check that the instance's modules haven't changed."""
+        self.test_runner.run_module_list_instance_after_reapply_md5()
+
+    @test(runs_after=[module_list_instance_after_reapply_md5])
+    def module_instances_after_reapply_md5(self):
+        """Check that the new instance still shows up in the list."""
+        self.test_runner.run_module_instances_after_reapply_md5()
+
+    @test(runs_after=[module_instances_after_reapply_md5])
+    def module_instance_count_after_reapply_md5(self):
+        """Check that the instance count hasn't changed."""
+        self.test_runner.run_module_instance_count_after_reapply_md5()
+
+    @test(runs_after=[module_instance_count_after_reapply_md5])
+    def module_reapply_all(self):
+        """Check that module reapply works."""
+        self.test_runner.run_module_reapply_all()
+
+    @test(runs_after=[module_reapply_all])
+    def module_reapply_all_wait(self):
+        """Wait for module reapply to complete."""
+        self.test_runner.run_module_reapply_all_wait()
+
+    @test(runs_after=[module_reapply_all_wait])
+    def module_instance_count_after_reapply(self):
+        """Check that the reapply instance count is right."""
+        self.test_runner.run_module_instance_count_after_reapply()
+
+    @test(runs_after=[module_instance_count_after_reapply])
+    def module_reapply_with_force(self):
+        """Check that module reapply with force works."""
+        self.test_runner.run_module_reapply_with_force()
+
+    @test(runs_after=[module_reapply_with_force])
+    def module_reapply_with_force_wait(self):
+        """Wait for module reapply with force to complete."""
+        self.test_runner.run_module_reapply_with_force_wait()
+
+    @test(runs_after=[module_reapply_with_force_wait])
+    def module_list_instance_after_reapply_force(self):
+        """Check that the new instance still has the right modules."""
+        self.test_runner.run_module_list_instance_after_reapply()
+
+    @test(runs_after=[module_list_instance_after_reapply_force])
+    def module_instances_after_reapply_force(self):
+        """Check that the new instance still shows up in the list."""
+        self.test_runner.run_module_instances_after_reapply()
+
+    @test(runs_after=[module_instances_after_reapply_force])
+    def module_instance_count_after_reapply_force(self):
+        """Check that the instance count is right after reapply force."""
+        self.test_runner.run_module_instance_count_after_reapply()
+
 
 @test(depends_on_groups=[groups.MODULE_INST_CREATE_WAIT],
       groups=[GROUP, groups.MODULE_INST, groups.MODULE_INST_DELETE])
@@ -548,6 +650,11 @@ class ModuleInstDeleteGroup(TestGroup):
         """Check that instance with module can be deleted."""
         self.test_runner.run_delete_inst_with_mods()
 
+    @test(runs_after=[delete_inst_with_mods])
+    def remove_mods_from_main_inst(self):
+        """Check that modules can be removed from the main instance."""
+        self.test_runner.run_remove_mods_from_main_inst()
+
 
 @test(depends_on_groups=[groups.MODULE_INST_DELETE],
       groups=[GROUP, groups.MODULE_INST, groups.MODULE_INST_DELETE_WAIT],

trove/tests/scenario/runners/module_runners.py

@@ -18,9 +18,12 @@ import Crypto.Random
 from proboscis import SkipTest
 import re
 import tempfile
+import time
 
 from troveclient.compat import exceptions
 
+from trove.common import exception
+from trove.common.utils import poll_until
 from trove.guestagent.common import guestagent_utils
 from trove.guestagent.common import operating_system
 from trove.module import models
@@ -64,8 +67,12 @@ class ModuleRunner(TestRunner):
             {'suffix': '_updated', 'priority': False, 'order': 8},
         ]
 
+        self.apply_count = 0
         self.mod_inst_id = None
+        self.mod_inst_apply_count = 0
         self.temp_module = None
+        self.live_update_orig_md5 = None
+        self.reapply_max_upd_date = None
         self._module_type = None
 
         self.test_modules = []
@@ -104,6 +111,10 @@ class ModuleRunner(TestRunner):
     def update_test_module(self):
         return self._get_test_module(1)
 
+    @property
+    def live_update_test_module(self):
+        return self._find_live_update_module()
+
     def build_module_args(self, name_order=None):
         suffix = "_unknown"
         priority = False
@@ -153,6 +164,11 @@ class ModuleRunner(TestRunner):
             return mod.auto_apply and mod.tenant_id and mod.visible
         return self._find_module(_match, "Could not find auto-apply module")
 
+    def _find_live_update_module(self):
+        def _match(mod):
+            return mod.live_update and mod.tenant_id and mod.visible
+        return self._find_module(_match, "Could not find live-update module")
+
     def _find_all_tenant_module(self):
         def _match(mod):
             return mod.tenant_id is None and mod.visible
@@ -584,6 +600,7 @@ class ModuleRunner(TestRunner):
         self.assert_module_create(
             self.admin_client, 5,
             live_update=True)
+        self.live_update_orig_md5 = self.test_modules[-1].md5
 
     def run_module_create_admin_priority_apply(self):
         self.assert_module_create(
@@ -905,10 +922,13 @@ class ModuleRunner(TestRunner):
         rowcount = len(instance_count_list)
         self.assert_equal(expected_rows, rowcount,
                           "Wrong number of instance count records from module")
-        if expected_rows == 1:
-            self.assert_equal(expected_count,
-                              instance_count_list[0].instance_count,
-                              "Wrong count in record from module instances")
+        # expected_count is a dict of md5->count pairs.
+        if expected_rows and expected_count:
+            for row in instance_count_list:
+                self.assert_equal(
+                    expected_count[row.module_md5], row.instance_count,
+                    "Wrong count in record from module instances; md5: %s" %
+                    row.module_md5)
 
     def run_module_query_empty(self):
         self.assert_module_query(
@@ -918,7 +938,7 @@ class ModuleRunner(TestRunner):
     def run_module_query_after_remove(self):
         self.assert_module_query(
             self.auth_client, self.instance_info.id,
-            self.module_auto_apply_count_prior_to_create + 1)
+            self.module_auto_apply_count_prior_to_create + 2)
 
     def assert_module_query(self, client, instance_id, expected_count,
                             expected_http_code=200, expected_results=None):
@@ -955,6 +975,7 @@ class ModuleRunner(TestRunner):
     def run_module_apply(self):
         self.assert_module_apply(self.auth_client, self.instance_info.id,
                                  self.main_test_module)
+        self.apply_count += 1
 
     def assert_module_apply(self, client, instance_id, module,
                             expected_is_admin=False,
@@ -1041,15 +1062,16 @@ class ModuleRunner(TestRunner):
 
     def run_module_list_instance_after_apply(self):
         self.assert_module_list_instance(
-            self.auth_client, self.instance_info.id, 1)
+            self.auth_client, self.instance_info.id, self.apply_count)
 
     def run_module_apply_another(self):
         self.assert_module_apply(self.auth_client, self.instance_info.id,
                                  self.update_test_module)
+        self.apply_count += 1
 
     def run_module_list_instance_after_apply_another(self):
         self.assert_module_list_instance(
-            self.auth_client, self.instance_info.id, 2)
+            self.auth_client, self.instance_info.id, self.apply_count)
 
     def run_module_update_after_remove(self):
         name, description, contents, priority, order = (
@@ -1068,10 +1090,11 @@ class ModuleRunner(TestRunner):
 
     def run_module_instance_count_after_apply(self):
         self.assert_module_instance_count(
-            self.auth_client, self.main_test_module.id, 1, 1)
+            self.auth_client, self.main_test_module.id, 1,
+            {self.main_test_module.md5: 1})
 
     def run_module_query_after_apply(self):
-        expected_count = self.module_auto_apply_count_prior_to_create + 1
+        expected_count = self.module_auto_apply_count_prior_to_create + 2
         expected_results = self.create_default_query_expected_results(
             [self.main_test_module])
         self.assert_module_query(self.auth_client, self.instance_info.id,
@@ -1110,16 +1133,44 @@ class ModuleRunner(TestRunner):
 
     def run_module_instance_count_after_apply_another(self):
         self.assert_module_instance_count(
-            self.auth_client, self.main_test_module.id, 1, 1)
+            self.auth_client, self.main_test_module.id, 1,
+            {self.main_test_module.md5: 1})
 
     def run_module_query_after_apply_another(self):
-        expected_count = self.module_auto_apply_count_prior_to_create + 2
+        expected_count = self.module_auto_apply_count_prior_to_create + 3
         expected_results = self.create_default_query_expected_results(
             [self.main_test_module, self.update_test_module])
         self.assert_module_query(self.auth_client, self.instance_info.id,
                                  expected_count=expected_count,
                                  expected_results=expected_results)
 
+    def run_module_update_not_live(
+            self, expected_exception=exceptions.Forbidden,
+            expected_http_code=403):
+        client = self.auth_client
+        self.assert_raises(
+            expected_exception, expected_http_code,
+            client, client.modules.update,
+            self.main_test_module.id, description='Do not allow this change')
+
+    def run_module_apply_live_update(self):
+        module = self.live_update_test_module
+        self.assert_module_apply(self.auth_client, self.instance_info.id,
+                                 module, expected_is_admin=module.is_admin)
+        self.apply_count += 1
+
+    def run_module_list_instance_after_apply_live(self):
+        self.assert_module_list_instance(
+            self.auth_client, self.instance_info.id, self.apply_count)
+
+    def run_module_update_live_update(self):
+        module = self.live_update_test_module
+        new_contents = self.get_module_contents(name=module.name + '_upd')
+        self.assert_module_update(
+            self.admin_client,
+            module.id,
+            contents=new_contents)
+
     def run_module_update_after_remove_again(self):
         self.assert_module_update(
             self.auth_client,
@@ -1129,10 +1180,13 @@ class ModuleRunner(TestRunner):
             all_datastore_versions=True)
 
     def run_create_inst_with_mods(self, expected_http_code=200):
+        live_update = self.live_update_test_module
         self.mod_inst_id = self.assert_inst_mod_create(
-            self.main_test_module.id, '_module', expected_http_code)
+            [self.main_test_module.id, live_update.id],
+            '_module', expected_http_code)
+        self.mod_inst_apply_count += 2
 
-    def assert_inst_mod_create(self, module_id, name_suffix,
+    def assert_inst_mod_create(self, module_ids, name_suffix,
                                expected_http_code):
         client = self.auth_client
         inst = client.instances.create(
@@ -1142,7 +1196,7 @@ class ModuleRunner(TestRunner):
             datastore=self.instance_info.dbaas_datastore,
             datastore_version=self.instance_info.dbaas_datastore_version,
             nics=self.instance_info.nics,
-            modules=[module_id],
+            modules=module_ids,
         )
         self.assert_client_code(client, expected_http_code)
         self.register_debug_inst_ids(inst.id)
@@ -1188,7 +1242,7 @@ class ModuleRunner(TestRunner):
 
     def run_module_query_after_inst_create(self):
         auto_modules = self._find_all_auto_apply_modules(visible=True)
-        expected_count = 1 + len(auto_modules)
+        expected_count = self.mod_inst_apply_count + len(auto_modules)
         expected_results = self.create_default_query_expected_results(
             [self.main_test_module] + auto_modules)
         self.assert_module_query(self.auth_client, self.mod_inst_id,
@@ -1197,7 +1251,7 @@ class ModuleRunner(TestRunner):
 
     def run_module_retrieve_after_inst_create(self):
         auto_modules = self._find_all_auto_apply_modules(visible=True)
-        expected_count = 1 + len(auto_modules)
+        expected_count = self.mod_inst_apply_count + len(auto_modules)
         expected_results = self.create_default_query_expected_results(
             [self.main_test_module] + auto_modules)
         self.assert_module_retrieve(self.auth_client, self.mod_inst_id,
@@ -1242,7 +1296,7 @@ class ModuleRunner(TestRunner):
 
     def run_module_query_after_inst_create_admin(self):
         auto_modules = self._find_all_auto_apply_modules()
-        expected_count = 1 + len(auto_modules)
+        expected_count = self.mod_inst_apply_count + len(auto_modules)
         expected_results = self.create_default_query_expected_results(
             [self.main_test_module] + auto_modules, is_admin=True)
         self.assert_module_query(self.admin_client, self.mod_inst_id,
@@ -1250,9 +1304,8 @@ class ModuleRunner(TestRunner):
                                  expected_results=expected_results)
 
     def run_module_retrieve_after_inst_create_admin(self):
-        pass
         auto_modules = self._find_all_auto_apply_modules()
-        expected_count = 1 + len(auto_modules)
+        expected_count = self.mod_inst_apply_count + len(auto_modules)
         expected_results = self.create_default_query_expected_results(
             [self.main_test_module] + auto_modules, is_admin=True)
         self.assert_module_retrieve(self.admin_client, self.mod_inst_id,
@@ -1268,6 +1321,143 @@ class ModuleRunner(TestRunner):
             expected_exception, expected_http_code,
             client, client.modules.delete, module.id)
 
+    def run_module_list_instance_after_mod_inst(self):
+        self.assert_module_list_instance(
+            self.auth_client, self.mod_inst_id,
+            self.module_auto_apply_create_count + 2)
+
+    def run_module_instances_after_mod_inst(self):
+        self.assert_module_instances(
+            self.auth_client, self.live_update_test_module.id, 2)
+
+    def run_module_instance_count_after_mod_inst(self):
+        self.assert_module_instance_count(
+            self.auth_client, self.live_update_test_module.id, 2,
+            {self.live_update_test_module.md5: 1,
+             self.live_update_orig_md5: 1})
+
+    def run_module_reapply_with_md5(self, expected_http_code=202):
+        self.assert_module_reapply(
+            self.auth_client, self.live_update_test_module,
+            expected_http_code=expected_http_code,
+            md5=self.live_update_test_module.md5)
+
+    def assert_module_reapply(self, client, module, expected_http_code,
+                              md5=None, force=False):
+        self.reapply_max_upd_date = self.get_updated(client, module.id)
+        client.modules.reapply(module.id, md5=md5, force=force)
+        self.assert_client_code(client, expected_http_code)
+
+    def run_module_reapply_with_md5_verify(self):
+        # since this isn't supposed to do anything, we can't 'wait' for it to
+        # finish, since we'll never know. So just sleep for a couple seconds
+        # just to make sure.
+        time.sleep(2)
+        # Now we check that the max_updated_date field didn't change
+        module_id = self.live_update_test_module.id
+        instance_count_list = self.auth_client.modules.instances(
+            module_id, count_only=True)
+        mismatch = False
+        for instance_count in instance_count_list:
+            if self.reapply_max_upd_date != instance_count.max_updated_date:
+                mismatch = True
+        self.assert_true(
+            mismatch,
+            "Could not find record having max_updated_date different from %s" %
+            self.reapply_max_upd_date)
+
+    def run_module_list_instance_after_reapply_md5(self):
+        self.assert_module_list_instance(
+            self.auth_client, self.mod_inst_id,
+            self.module_auto_apply_create_count + 2)
+
+    def run_module_instances_after_reapply_md5(self):
+        self.assert_module_instances(
+            self.auth_client, self.live_update_test_module.id, 2)
+
+    def run_module_instance_count_after_reapply_md5(self):
+        self.assert_module_instance_count(
+            self.auth_client, self.live_update_test_module.id, 2,
+            {self.live_update_test_module.md5: 1,
+             self.live_update_orig_md5: 1})
+
+    def run_module_reapply_all(self, expected_http_code=202):
+        module_id = self.live_update_test_module.id
+        client = self.auth_client
+        self.reapply_max_upd_date = self.get_updated(client, module_id)
+        self.assert_module_reapply(
+            client, self.live_update_test_module,
+            expected_http_code=expected_http_code)
+
+    def run_module_reapply_all_wait(self):
+        self.wait_for_reapply(
+            self.auth_client, self.live_update_test_module.id,
+            md5=self.live_update_orig_md5)
+
+    def wait_for_reapply(self, client, module_id, updated=None, md5=None):
+        """Reapply is done when all the counts for 'md5' are gone.  If updated
+        is passed in, the min_updated_date must all be greater than it.
+        """
+        if not updated and not md5:
+            raise RuntimeError("Code error: Must pass in 'updated' or 'md5'.")
+        self.report.log("Waiting for all md5:%s modules to have an updated "
+                        "date greater than %s" % (md5, updated))
+
+        def _all_updated():
+            min_updated = self.get_updated(
+                client, module_id, max=False, md5=md5)
+            if md5:
+                return min_updated is None
+            return min_updated > updated
+
+        timeout = 60
+        try:
+            poll_until(_all_updated, time_out=timeout, sleep_time=5)
+            self.report.log("All instances now have the current module "
+                            "for md5: %s." % md5)
+        except exception.PollTimeOut:
+            self.fail("Some instances were not updated with the "
+                      "timeout: %ds" % timeout)
+
+    def get_updated(self, client, module_id, max=True, md5=None):
+        updated = None
+        instance_count_list = client.modules.instances(
+            module_id, count_only=True)
+        for instance_count in instance_count_list:
+            if not md5 or md5 == instance_count.module_md5:
+                if not updated or (
+                        (max and instance_count.max_updated_date > updated) or
+                        (not max and
+                         instance_count.min_updated_date < updated)):
+                    updated = (instance_count.max_updated_date
+                               if max else instance_count.min_updated_date)
+        return updated
+
+    def run_module_list_instance_after_reapply(self):
+        self.assert_module_list_instance(
+            self.auth_client, self.mod_inst_id,
+            self.module_auto_apply_create_count + 2)
+
+    def run_module_instances_after_reapply(self):
+        self.assert_module_instances(
+            self.auth_client, self.live_update_test_module.id, 2)
+
+    def run_module_instance_count_after_reapply(self):
+        self.assert_module_instance_count(
+            self.auth_client, self.live_update_test_module.id, 1,
+            {self.live_update_test_module.md5: 2})
+
+    def run_module_reapply_with_force(self, expected_http_code=202):
+        self.assert_module_reapply(
+            self.auth_client, self.live_update_test_module,
+            expected_http_code=expected_http_code,
+            force=True)
+
+    def run_module_reapply_with_force_wait(self):
+        self.wait_for_reapply(
+            self.auth_client, self.live_update_test_module.id,
+            updated=self.reapply_max_upd_date)
+
     def run_delete_inst_with_mods(self, expected_http_code=202):
         self.assert_delete_instance(self.mod_inst_id, expected_http_code)
 
@@ -1276,6 +1466,14 @@ class ModuleRunner(TestRunner):
         client.instances.delete(instance_id)
         self.assert_client_code(client, expected_http_code)
 
+    def run_remove_mods_from_main_inst(self, expected_http_code=200):
+        client = self.auth_client
+        modquery_list = client.instances.module_query(self.instance_info.id)
+        self.assert_client_code(client, expected_http_code)
+        for modquery in modquery_list:
+            client.instances.module_remove(self.instance_info.id, modquery.id)
+            self.assert_client_code(client, expected_http_code)
+
     def run_wait_for_delete_inst_with_mods(
             self, expected_last_state=['SHUTDOWN']):
         self.assert_all_gone(self.mod_inst_id, expected_last_state)