def action(self, req, body, tenant_id, id):
        LOG.info("req : '%s'\n\n" % req)
        LOG.info("Committing an ACTION against instance %s for tenant '%s'"
                 % (id, tenant_id))
        if not body:
            raise exception.BadRequest(_("Invalid request body."))
        context = req.environ[wsgi.CONTEXT_KEY]
        instance = models.MgmtInstance.load(context=context, id=id)
        _actions = {
            'stop': self._action_stop,
            'reboot': self._action_reboot,
            'migrate': self._action_migrate,
            'reset-task-status': self._action_reset_task_status
        }
        selected_action = None
        for key in body:
            if key in _actions:
                if selected_action is not None:
                    msg = _("Only one action can be specified per request.")
                    raise exception.BadRequest(msg)
                selected_action = _actions[key]
            else:
                msg = _("Invalid instance action: %s") % key
                raise exception.BadRequest(msg)

        if selected_action:
            return selected_action(context, instance, body)
        else:
            raise exception.BadRequest(_("Invalid request body."))

    def _service_is_active(self):
        """
        Check that the database guest is active.

        This function is meant to be called with poll_until to check that
        the guest is alive before sending a 'create' message. This prevents
        over billing a customer for a instance that they can never use.

        Returns: boolean if the service is active.
        Raises: TroveError if the service is in a failure state.
        """
        service = InstanceServiceStatus.find_by(instance_id=self.id)
        status = service.get_status()
        if status == rd_instance.ServiceStatuses.RUNNING:
            return True
        elif status not in [rd_instance.ServiceStatuses.NEW,
                            rd_instance.ServiceStatuses.BUILDING]:
            raise TroveError(_("Service not active, status: %s") % status)

        c_id = self.db_info.compute_instance_id
        nova_status = self.nova_client.servers.get(c_id).status
        if nova_status in [InstanceStatus.ERROR,
                           InstanceStatus.FAILED]:
            raise TroveError(_("Server not active, status: %s") % nova_status)
        return False

    def reboot(self):
        try:
            LOG.debug(_("Instance %s calling stop_db...") % self.id)
            self.guest.stop_db()
            LOG.debug(_("Rebooting instance %s") % self.id)
            self.server.reboot()

            # Poll nova until instance is active
            reboot_time_out = CONF.reboot_time_out

            def update_server_info():
                self._refresh_compute_server_info()
                return self.server.status == 'ACTIVE'

            utils.poll_until(
                update_server_info,
                sleep_time=2,
                time_out=reboot_time_out)

            # Set the status to PAUSED. The guest agent will reset the status
            # when the reboot completes and MySQL is running.
            self._set_service_status_to_paused()
            LOG.debug(_("Successfully rebooted instance %s") % self.id)
        except Exception as e:
            LOG.error(_("Failed to reboot instance %(id)s: %(e)s") %
                      {'id': self.id, 'e': str(e)})
        finally:
            LOG.debug(_("Rebooting FINALLY  %s") % self.id)
            self.update_db(task_status=inst_models.InstanceTasks.NONE)

    def _callback_handler(self, message, callback):
        """Call callback with deserialized message.

        Messages that are processed without exception are ack'ed.

        If the message processing generates an exception, it will be
        ack'ed if ack_on_error=True. Otherwise it will be .reject()'ed.
        Rejection is better than waiting for the message to timeout.
        Rejected messages are immediately requeued.
        """

        ack_msg = False
        try:
            msg = rpc_common.deserialize_msg(message.payload)
            callback(msg)
            ack_msg = True
        except Exception:
            if self.ack_on_error:
                ack_msg = True
                LOG.exception(_("Failed to process message" " ... skipping it."))
            else:
                LOG.exception(_("Failed to process message" " ... will requeue."))
        finally:
            if ack_msg:
                message.ack()
            else:
                message.reject()

 def _error_callback(exc):
     if isinstance(exc, socket.timeout):
         LOG.debug(_("Timed out waiting for RPC response: %s") % str(exc))
         raise rpc_common.Timeout()
     else:
         LOG.exception(_("Failed to consume message from queue: %s") % str(exc))
         info["do_consume"] = True

    def _load_servers_status(load_instance, context, db_items, find_server):
        ret = []
        for db in db_items:
            server = None
            try:
                #TODO(tim.simpson): Delete when we get notifications working!
                if InstanceTasks.BUILDING == db.task_status:
                    db.server_status = "BUILD"
                else:
                    try:
                        server = find_server(db.id, db.compute_instance_id)
                        db.server_status = server.status
                    except exception.ComputeInstanceNotFound:
                        db.server_status = "SHUTDOWN"  # Fake it...
                #TODO(tim.simpson): End of hack.

                #volumes = find_volumes(server.id)
                datastore_status = InstanceServiceStatus.find_by(
                    instance_id=db.id)
                if not datastore_status.status:  # This should never happen.
                    LOG.error(_("Server status could not be read for "
                                "instance id(%s)") % db.id)
                    continue
                LOG.info(_("Server api_status(%s)") %
                         datastore_status.status.api_status)
            except exception.ModelNotFoundError:
                LOG.error(_("Server status could not be read for "
                            "instance id(%s)") % db.id)
                continue
            ret.append(load_instance(context, db, datastore_status,
                                     server=server))
        return ret

    def start_db(self, update_db=False):
        LOG.info(_("Starting MongoDB"))

        self._enable_db_on_boot()

        try:
            mongo_service = self._get_service()
            utils.execute_with_timeout(mongo_service['cmd_start'],
                                       shell=True)
        except ProcessExecutionError:
            pass
        except KeyError:
            raise RuntimeError("MongoDB service is not discovered.")

        if not self.status.wait_for_real_status_to_change_to(
                ds_instance.ServiceStatuses.RUNNING,
                self.state_change_wait_time, update_db):
            LOG.error(_("Start up of MongoDB failed"))
            # If it won't start, but won't die either, kill it by hand so we
            # don't let a rouge process wander around.
            try:
                out, err = utils.execute_with_timeout(
                    system.FIND_PID, shell=True)
                pid = "".join(out.split(" ")[1:2])
                utils.execute_with_timeout(
                    system.MONGODB_KILL % pid, shell=True)
            except exception.ProcessExecutionError as p:
                LOG.error("Error killing stalled MongoDB start command.")
                LOG.error(p)
                # There's nothing more we can do...
            self.status.end_install_or_restart()
            raise RuntimeError("Could not start MongoDB")

 def process_request(self, request):
     roles = request.headers.get('X_ROLE', '').split(',')
     LOG.debug(_("Processing auth request with roles: %s") % roles)
     tenant_id = request.headers.get('X-Tenant-Id', None)
     LOG.debug(_("Processing auth request with tenant_id: %s") % tenant_id)
     for provider in self.auth_providers:
         provider.authorize(request, tenant_id, roles)

    def execute_restore(self, context, backup_info, restore_location):

        try:
            LOG.debug("Getting Restore Runner %(type)s", backup_info)
            restore_runner = self._get_restore_runner(backup_info['type'])

            LOG.debug("Getting Storage Strategy")
            storage = get_storage_strategy(
                CONF.storage_strategy,
                CONF.storage_namespace)(context)

            runner = restore_runner(storage, location=backup_info['location'],
                                    checksum=backup_info['checksum'],
                                    restore_location=restore_location)
            backup_info['restore_location'] = restore_location
            LOG.debug("Restoring instance from backup %(id)s to "
                      "%(restore_location)s" % backup_info)
            content_size = runner.restore()
            LOG.info(_("Restore from backup %(id)s completed successfully "
                       "to %(restore_location)s") % backup_info)
            LOG.info(_("Restore size: %s") % content_size)

        except Exception as e:
            LOG.error(e)
            LOG.error(_("Error restoring backup %(id)s") % backup_info)
            raise

        else:
            LOG.info(_("Restored Backup %(id)s") % backup_info)

 def prepare(self, context, packages, databases, memory_mb, users,
             device_path=None, mount_point=None, backup_info=None,
             config_contents=None, root_password=None, overrides=None):
     """
     This is called when the trove instance first comes online.
     It is the first rpc message passed from the task manager.
     prepare handles all the base configuration of the redis instance.
     """
     try:
         app = RedisApp(RedisAppStatus.get())
         RedisAppStatus.get().begin_install()
         if device_path:
             device = volume.VolumeDevice(device_path)
             # unmount if device is already mounted
             device.unmount_device(device_path)
             device.format()
             device.mount(mount_point)
             operating_system.update_owner('redis', 'redis', mount_point)
             LOG.debug('Mounted the volume.')
         app.install_if_needed(packages)
         LOG.info(_('Securing redis now.'))
         app.write_config(config_contents)
         app.restart()
         LOG.info(_('"prepare" redis call has finished.'))
     except Exception as e:
         LOG.error(e)
         app.status.set_status(rd_instance.ServiceStatuses.FAILED)
         raise RuntimeError("prepare call has failed.")

 def index(self, req, tenant_id, detailed=False):
     """Return all hosts."""
     LOG.info(_("req : '%s'\n\n") % req)
     LOG.info(_("Indexing a host for tenant '%s'") % tenant_id)
     context = req.environ[wsgi.CONTEXT_KEY]
     hosts = models.SimpleHost.load_all(context)
     return wsgi.Result(views.HostsView(hosts).data(), 200)

        def _inner():
            if initial_delay:
                greenthread.sleep(initial_delay)

            try:
                while self._running:
                    start = timeutils.utcnow()
                    self.f(*self.args, **self.kw)
                    end = timeutils.utcnow()
                    if not self._running:
                        break
                    delay = interval - timeutils.delta_seconds(start, end)
                    if delay <= 0:
                        LOG.warn(_('task run outlasted interval by %s sec') %
                                 -delay)
                    greenthread.sleep(delay if delay > 0 else 0)
            except LoopingCallDone as e:
                self.stop()
                done.send(e.retvalue)
            except Exception:
                LOG.exception(_('in fixed duration looping call'))
                done.send_exception(*sys.exc_info())
                return
            else:
                done.send(True)

 def pkg_version(self, package_name):
     cmd_list = ["dpkg", "-l", package_name]
     p = commands.getstatusoutput(" ".join(cmd_list))
     # check the command status code
     if not p[0] == 0:
         return None
     # Need to capture the version string
     # check the command output
     std_out = p[1]
     patterns = [".*No packages found matching.*", "\w\w\s+(\S+)\s+(\S+)\s+(.*)$"]
     for line in std_out.split("\n"):
         for p in patterns:
             regex = re.compile(p)
             matches = regex.match(line)
             if matches:
                 line = matches.group()
                 parts = line.split()
                 if not parts:
                     msg = _("returned nothing")
                     LOG.error(msg)
                     raise exception.GuestError(msg)
                 if len(parts) <= 2:
                     msg = _("Unexpected output.")
                     LOG.error(msg)
                     raise exception.GuestError(msg)
                 if parts[1] != package_name:
                     msg = _("Unexpected output:[1] = %s") % str(parts[1])
                     LOG.error(msg)
                     raise exception.GuestError(msg)
                 if parts[0] == "un" or parts[2] == "<none>":
                     return None
                 return parts[2]
     msg = _("version() saw unexpected output from dpkg!")
     LOG.error(msg)

 def update_all(self, req, body, tenant_id, instance_id):
     """Change the password of one or more users."""
     LOG.info(_("Updating user passwords for instance '%s'") % instance_id)
     LOG.info(logging.mask_password(_("req : '%s'\n\n") % req))
     context = req.environ[wsgi.CONTEXT_KEY]
     users = body['users']
     model_users = []
     for user in users:
         try:
             mu = guest_models.MySQLUser()
             mu.name = user['name']
             mu.host = user.get('host')
             mu.password = user['password']
             found_user = models.User.load(context, instance_id,
                                           mu.name, mu.host)
             if not found_user:
                 user_and_host = mu.name
                 if mu.host:
                     user_and_host += '@' + mu.host
                 raise exception.UserNotFound(uuid=user_and_host)
             model_users.append(mu)
         except (ValueError, AttributeError) as e:
             raise exception.BadRequest(msg=str(e))
     models.User.change_password(context, instance_id, model_users)
     return wsgi.Result(None, 202)

 def _delete_resources():
     if self.is_building:
         raise exception.UnprocessableEntity("Instance %s is not ready." % self.id)
     LOG.debug(_("  ... deleting compute id = %s") % self.db_info.compute_instance_id)
     LOG.debug(_(" ... setting status to DELETING."))
     self.update_db(task_status=InstanceTasks.DELETING)
     task_api.API(self.context).delete_instance(self.id)

 def update_owner(self, path):
     LOG.info(_("Set owner to 'mongodb' for %s ") % system.CONFIG)
     utils.execute_with_timeout("chown", "-R", "mongodb", path,
                                run_as_root=True, root_helper="sudo")
     LOG.info(_("Set group to 'mongodb' for %s ") % system.CONFIG)
     utils.execute_with_timeout("chgrp", "-R", "mongodb", path,
                                run_as_root=True, root_helper="sudo")

    def start_mysql(self, update_db=False):
        LOG.info(_("Starting mysql..."))
        # This is the site of all the trouble in the restart tests.
        # Essentially what happens is that mysql start fails, but does not
        # die. It is then impossible to kill the original, so

        self._enable_mysql_on_boot()

        try:
            mysql_service = operating_system.service_discovery(
                MYSQL_SERVICE_CANDIDATES)
            utils.execute_with_timeout(mysql_service['cmd_start'], shell=True)
        except KeyError:
            raise RuntimeError("Service is not discovered.")
        except exception.ProcessExecutionError:
            # it seems mysql (percona, at least) might come back with [Fail]
            # but actually come up ok. we're looking into the timing issue on
            # parallel, but for now, we'd like to give it one more chance to
            # come up. so regardless of the execute_with_timeout() response,
            # we'll assume mysql comes up and check it's status for a while.
            pass
        if not self.status.wait_for_real_status_to_change_to(
                rd_instance.ServiceStatuses.RUNNING,
                self.state_change_wait_time, update_db):
            LOG.error(_("Start up of MySQL failed!"))
            # If it won't start, but won't die either, kill it by hand so we
            # don't let a rouge process wander around.
            try:
                utils.execute_with_timeout("sudo", "pkill", "-9", "mysql")
            except exception.ProcessExecutionError as p:
                LOG.error("Error killing stalled mysql start command.")
                LOG.error(p)
                # There's nothing more we can do...
            self.status.end_install_or_restart()
            raise RuntimeError("Could not start MySQL!")

 def index(self, req, tenant_id):
     """Return all storage devices."""
     LOG.info(_("req : '%s'\n\n") % req)
     LOG.info(_("Indexing storage info for tenant '%s'") % tenant_id)
     context = req.environ[wsgi.CONTEXT_KEY]
     storages = models.StorageDevices.load(context)
     return wsgi.Result(views.StoragesView(storages).data(), 200)

 def show(self, req, tenant_id, id):
     """Return a single backup."""
     LOG.info(_("Showing a backup for tenant '%s'") % tenant_id)
     LOG.info(_("id : '%s'\n\n") % id)
     context = req.environ[wsgi.CONTEXT_KEY]
     backup = Backup.get_by_id(context, id)
     return wsgi.Result(views.BackupView(backup).data(), 200)

 def mysql_is_running(self):
     if base.exec_with_root_helper("/usr/bin/mysqladmin", "ping"):
         LOG.info(_("The mysqld daemon is up and running."))
         return True
     else:
         LOG.info(_("The mysqld daemon is not running."))
         return False