Merge "Add support for module-reapply command"

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)