def create(self, req, body, tenant_id, version_id):
        """Create configuration parameter for datastore version."""
        LOG.info(_("Creating configuration parameter for datastore"))
        LOG.debug("req : '%s'\n\n" % req)
        LOG.debug("body : '%s'\n\n" % body)
        if not body:
            raise exception.BadRequest(_("Invalid request body."))

        parameter = body['configuration-parameter']
        name = parameter['name']
        restart_required = bool(parameter['restart_required'])
        data_type, min_size, max_size = self._validate_data_type(parameter)
        datastore_version = ds_models.DatastoreVersion.load_by_uuid(version_id)

        rule = config_models.DatastoreConfigurationParameters.create(
            name=name,
            datastore_version_id=datastore_version.id,
            restart_required=restart_required,
            data_type=data_type,
            max_size=max_size,
            min_size=min_size
        )
        return wsgi.Result(
            views.MgmtConfigurationParameterView(rule).data(),
            200)

def chown(path, user, group, recursive=True, force=False, **kwargs):
    """Changes the owner and group of a given file.

    seealso:: _execute_shell_cmd for valid optional keyword arguments.

    :param path:         Path to the modified file.
    :type path:          string

    :param user:         Owner.
    :type user:          string

    :param group:        Group.
    :type group:         string

    :param recursive:    Operate on files and directories recursively.
    :type recursive:     boolean

    :param force:        Suppress most error messages.
    :type force:         boolean

    :raises:             :class:`UnprocessableEntity` if path not given.
    :raises:             :class:`UnprocessableEntity` if owner/group not given.
    """

    if not path:
        raise exception.UnprocessableEntity(
            _("Cannot change ownership of a blank file or directory."))
    if not user and not group:
        raise exception.UnprocessableEntity(
            _("Please specify owner or group, or both."))

    owner_group_modifier = _build_user_group_pair(user, group)
    options = (('f', force), ('R', recursive))
    _execute_shell_cmd('chown', options, owner_group_modifier, path, **kwargs)

def move(source, destination, force=False, **kwargs):
    """Move a given file or directory to a new location.
    Move attempts to preserve the original ownership, permissions and
    timestamps.

    :seealso: _execute_shell_cmd for valid optional keyword arguments.

    :param source:          Path to the source location.
    :type source:           string

    :param destination:     Path to the destination location.
    :type destination:      string

    :param force:           Do not prompt before overwriting.
    :type force:            boolean

    :raises:                :class:`UnprocessableEntity` if source or
                            destination not given.
    """

    if not source:
        raise exception.UnprocessableEntity(_("Missing source path."))
    elif not destination:
        raise exception.UnprocessableEntity(_("Missing destination path."))

    options = (('f', force),)
    _execute_shell_cmd('mv', options, source, destination, **kwargs)

    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 clear_expired_password():
    """
    Some mysql installations generate random root password
    and save it in /root/.mysql_secret, this password is
    expired and should be changed by client that supports expired passwords.
    """
    LOG.debug("Removing expired password.")
    secret_file = "/root/.mysql_secret"
    try:
        out, err = utils.execute("cat", secret_file,
                                 run_as_root=True, root_helper="sudo")
    except exception.ProcessExecutionError:
        LOG.exception(_("/root/.mysql_secret does not exist."))
        return
    m = re.match('# The random password set for the root user at .*: (.*)',
                 out)
    if m:
        try:
            out, err = utils.execute("mysqladmin", "-p%s" % m.group(1),
                                     "password", "", run_as_root=True,
                                     root_helper="sudo")
        except exception.ProcessExecutionError:
            LOG.exception(_("Cannot change mysql password."))
            return
        operating_system.remove(secret_file, force=True, as_root=True)
        LOG.debug("Expired password removed.")

    def write_config(
        self,
        config_contents,
        execute_function=utils.execute_with_timeout,
        mkstemp_function=tempfile.mkstemp,
        unlink_function=os.unlink,
    ):

        # first securely create a temp file. mkstemp() will set
        # os.O_EXCL on the open() call, and we get a file with
        # permissions of 600 by default.
        (conf_fd, conf_path) = mkstemp_function()

        LOG.debug("Storing temporary configuration at %s." % conf_path)

        # write config and close the file, delete it if there is an
        # error. only unlink if there is a problem. In normal course,
        # we move the file.
        try:
            os.write(conf_fd, config_contents)
            execute_function("sudo", "mv", conf_path, system.CASSANDRA_CONF)
            # TODO(denis_makogon): figure out the dynamic way to discover
            # configs owner since it can cause errors if there is
            # no cassandra user in operating system
            execute_function("sudo", "chown", "cassandra:cassandra", system.CASSANDRA_CONF)
            execute_function("sudo", "chmod", "a+r", system.CASSANDRA_CONF)
        except Exception:
            LOG.exception(_("Exception generating Cassandra configuration %s.") % conf_path)
            unlink_function(conf_path)
            raise
        finally:
            os.close(conf_fd)

        LOG.info(_("Wrote new Cassandra configuration."))

        def _shrink_cluster():
            db_instances = DBInstance.find_all(cluster_id=cluster_id,
                                               deleted=False).all()

            all_instance_ids = [db_instance.id for db_instance in db_instances]

            remove_instances = [Instance.load(context, instance_id)
                                for instance_id in instance_ids]

            left_instances = [Instance.load(context, instance_id)
                              for instance_id
                              in all_instance_ids
                              if instance_id not in instance_ids]

            remove_member_ips = [self.get_ip(instance)
                                 for instance in remove_instances]

            k = VerticaCluster.k_safety(len(left_instances))

            for db_instance in db_instances:
                if db_instance['type'] == 'master':
                    master_instance = Instance.load(context,
                                                    db_instance.id)
                    if self.get_ip(master_instance) in remove_member_ips:
                        raise RuntimeError(_("Cannot remove master instance!"))
                    LOG.debug(_("Marking cluster k-safety: %s") % k)
                    self.get_guest(master_instance).mark_design_ksafe(k)
                    self.get_guest(master_instance).shrink_cluster(
                        remove_member_ips)
                    break

            for r in remove_instances:
                Instance.delete(r)

    def stop_db_service(self, service_candidates, timeout,
                        disable_on_boot=False, update_db=False):
        """Stop the database service and wait for the database to shutdown.

        :param service_candidates:   List of possible system service names.
        :type service_candidates:    list

        :param timeout:              Wait timeout in seconds.
        :type timeout:               integer

        :param disable_on_boot:      Disable service auto-start.
                                     The auto-start setting will be updated
                                     only if the service command succeeds.
        :type disable_on_boot:       boolean

        :param update_db:            Suppress the Trove instance heartbeat.
        :type update_db:             boolean

        :raises:              :class:`RuntimeError` on failure.
        """
        LOG.info(_("Stopping database service."))
        operating_system.stop_service(service_candidates)

        LOG.debug("Waiting for database to shutdown.")
        if not self._wait_for_database_service_status(
                instance.ServiceStatuses.SHUTDOWN, timeout, update_db):
            raise RuntimeError(_("Database failed to stop."))

        LOG.info(_("Database has stopped successfully."))

        if disable_on_boot:
            LOG.info(_("Disable service auto-start on boot."))
            operating_system.disable_service_on_boot(service_candidates)

 def _run_pre_backup(self):
     try:
         for cmd in self.pre_backup_commands:
             utils.execute_with_timeout(*cmd)
         root = service.CouchbaseRootAccess()
         pw = root.get_password()
         self._save_buckets_config(pw)
         with open(OUTFILE, "r") as f:
             out = f.read()
             if out != "[]":
                 d = json.loads(out)
                 all_memcached = True
                 for i in range(len(d)):
                     bucket_type = d[i]["bucketType"]
                     if bucket_type != "memcached":
                         all_memcached = False
                         break
                 if not all_memcached:
                     self._backup(pw)
                 else:
                     LOG.info(_("All buckets are memcached.  "
                                "Skipping backup."))
         operating_system.move(OUTFILE, system.COUCHBASE_DUMP_DIR)
         if pw != "password":
             # Not default password, backup generated root password
             operating_system.copy(system.pwd_file,
                                   system.COUCHBASE_DUMP_DIR,
                                   preserve=True, as_root=True)
     except exception.ProcessExecutionError:
         LOG.exception(_("Error during pre-backup phase."))
         raise

 def _enable_db_on_boot(self):
     LOG.info(_("Enabling MongoDB on boot."))
     try:
         mongo_service = self._get_service()
         utils.execute_with_timeout(mongo_service["cmd_enable"], shell=True)
     except KeyError:
         raise RuntimeError(_("MongoDB service is not discovered."))

    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:
                LOG.exception(_("Error killing stalled MySQL start command."))
                # There's nothing more we can do...
            self.status.end_install_or_restart()
            raise RuntimeError("Could not start MySQL!")

    def _wait_for_slave_status(self, status, client, max_time):

        def verify_slave_status():
            actual_status = client.execute(
                "SHOW GLOBAL STATUS like 'slave_running'").first()
            if actual_status:
                return actual_status[1].upper() == status.upper()
            # The slave_running status is no longer available in MySql 5.7
            # Need to query the performance_schema instead.
            LOG.debug("slave_running global status doesn't exist, checking "
                      "service_state in performance_schema instead.")
            q = sql_query.Query()
            q.columns = ["a.service_state", "c.service_state"]
            q.tables = ["performance_schema.replication_applier_status a",
                        "performance_schema.replication_connection_status c"]
            q.where = ["a.channel_name = ''", "c.channel_name = ''"]
            t = text(str(q))
            actual_status = client.execute(t).first()
            if (actual_status and actual_status[0].upper() == 'ON' and
                    actual_status[1].upper() == 'ON'):
                actual_status_str = 'ON'
            else:
                actual_status_str = 'OFF'
            return actual_status_str == status.upper()

        LOG.debug("Waiting for SLAVE_RUNNING to change to %s.", status)
        try:
            utils.poll_until(verify_slave_status, sleep_time=3,
                             time_out=max_time)
            LOG.info(_("Replication is now %s.") % status.lower())
        except PollTimeOut:
            raise RuntimeError(
                _("Replication is not %(status)s after %(max)d seconds.") % {
                    'status': status.lower(), 'max': max_time})

 def _disable_mysql_on_boot(self):
     try:
         utils.execute_with_timeout(self.mysql_service['cmd_disable'],
                                    shell=True)
     except KeyError:
         LOG.exception(_("Error disabling MySQL start on boot."))
         raise RuntimeError(_("Service is not discovered."))

 def _get_user(self, username, hostname):
     """Return a single user matching the criteria."""
     user = None
     try:
         # Could possibly throw a ValueError here.
         user = models.MySQLUser(name=username)
         user.check_reserved()
         if username == ADMIN_USER_NAME and hostname == ADMIN_HOST:
             raise ValueError(
                 "User %[email protected]%s is reserved." % (ADMIN_USER_NAME, ADMIN_HOST))
     except ValueError as ve:
         LOG.exception(_("Error Getting user information"))
         err_msg = encodeutils.exception_to_unicode(ve)
         raise exception.BadRequest(_("Username %(user)s is not valid"
                                      ": %(reason)s") %
                                    {'user': username, 'reason': err_msg}
                                    )
     with self.local_sql_client(self.mysql_app.get_engine()) as client:
         q = sql_query.Query()
         q.columns = ['User', 'Host']
         q.tables = ['mysql.user']
         q.where = ["Host != 'localhost'",
                    "User = '%s'" % username,
                    "Host = '%s'" % hostname]
         q.order = ['User', 'Host']
         t = text(str(q))
         result = client.execute(t).fetchall()
         LOG.debug("Getting user information %s." % result)
         if len(result) != 1:
             return None
         found_user = result[0]
         user.host = found_user['Host']
         self._associate_dbs(user)
         return user

    def create_user(self, users):
        LOG.debug("Creating user(s) for accessing DB2 database(s).")
        try:
            for item in users:
                user = models.DatastoreUser.deserialize(item)
                user.check_create()
                try:
                    LOG.debug("Creating OS user: %s." % user.name)
                    utils.execute_with_timeout(
                        system.CREATE_USER_COMMAND % {
                            'login': user.name, 'login': user.name,
                            'passwd': user.password}, shell=True)
                except exception.ProcessExecutionError as pe:
                    LOG.exception(_("Error creating user: %s.") % user.name)
                    continue

                for database in user.databases:
                    mydb = models.DatastoreSchema.deserialize(database)
                    try:
                        LOG.debug("Granting user: %s access to database: %s."
                                  % (user.name, mydb.name))
                        run_command(system.GRANT_USER_ACCESS % {
                            'dbname': mydb.name, 'login': user.name})
                    except exception.ProcessExecutionError as pe:
                        LOG.debug(
                            "Error granting user: %s access to database: %s."
                            % (user.name, mydb.name))
                        LOG.debug(pe)
                        pass
        except exception.ProcessExecutionError as pe:
            LOG.exception(_("An error occurred creating users: %s.") %
                          pe.message)
            pass

 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 delete_user(self, user):
        LOG.debug("Delete a given user.")
        db2_user = models.DatastoreUser.deserialize(user)
        db2_user.check_delete()
        userName = db2_user.name
        user_dbs = db2_user.databases
        LOG.debug("For user %s, databases to be deleted = %r." % (
            userName, user_dbs))

        if len(user_dbs) == 0:
            databases = self.list_access(db2_user.name, None)
        else:
            databases = user_dbs

        LOG.debug("databases for user = %r." % databases)
        for database in databases:
            mydb = models.DatastoreSchema.deserialize(database)
            try:
                run_command(system.REVOKE_USER_ACCESS % {
                    'dbname': mydb.name,
                    'login': userName})
                LOG.debug("Revoked access for user:%s on database:%s." % (
                    userName, mydb.name))
            except exception.ProcessExecutionError as pe:
                LOG.debug("Error occurred while revoking access to %s." %
                          mydb.name)
                pass
            try:
                utils.execute_with_timeout(system.DELETE_USER_COMMAND % {
                    'login': db2_user.name.lower()}, shell=True)
            except exception.ProcessExecutionError as pe:
                LOG.exception(_(
                    "There was an error while deleting user: %s.") % pe)
                raise exception.GuestError(original_message=_(
                    "Unable to delete user: %s.") % userName)

 def action(self, req, body, tenant_id, id):
     LOG.debug(("Committing Action Against Cluster for "
                "Tenant '%(tenant_id)s'\n"
                "req : '%(req)s'\n\nid : '%(id)s'\n\n") %
               {"req": req, "id": id, "tenant_id": tenant_id})
     if not body:
         raise exception.BadRequest(_("Invalid request body."))
     context = req.environ[wsgi.CONTEXT_KEY]
     cluster = models.Cluster.load(context, id)
     manager = cluster.datastore_version.manager
     api_strategy = strategy.load_api_strategy(manager)
     _actions = api_strategy.cluster_controller_actions
     selected_action = None
     for key in body:
         if key in _actions:
             selected_action = _actions[key]
             break
     else:
         message = _("No action '%(action)s' supplied "
                     "by strategy for manager '%(manager)s'") % (
                         {'action': key, 'manager': manager})
         raise exception.TroveError(message)
     cluster = selected_action(cluster, body)
     if cluster:
         view = views.load_view(cluster, req=req, load_servers=False)
         wsgi_result = wsgi.Result(view.data(), 202)
     else:
         wsgi_result = wsgi.Result(None, 202)
     return wsgi_result

    def grow_cluster(self, context, cluster_id, new_instance_ids):

        def _grow_cluster():
            LOG.debug("begin grow_cluster for Vertica cluster %s" % cluster_id)

            db_instances = DBInstance.find_all(cluster_id=cluster_id,
                                               deleted=False).all()

            instance_ids = [db_instance.id for db_instance in db_instances]

            # Wait for new cluster members to get to cluster-ready status.
            if not self._all_instances_ready(new_instance_ids, cluster_id):
                return

            new_insts = [Instance.load(context, instance_id)
                         for instance_id in new_instance_ids]

            existing_instances = [Instance.load(context, instance_id)
                                  for instance_id
                                  in instance_ids
                                  if instance_id not in new_instance_ids]

            existing_guests = [self.get_guest(i) for i in existing_instances]
            new_guests = [self.get_guest(i) for i in new_insts]
            all_guests = new_guests + existing_guests

            authorized_users_without_password = ['root', 'dbadmin']
            new_ips = [self.get_ip(instance) for instance in new_insts]

            for user in authorized_users_without_password:
                pub_key = [guest.get_public_keys(user) for guest in all_guests]
                for guest in all_guests:
                    guest.authorize_public_keys(user, pub_key)

            for db_instance in db_instances:
                if db_instance['type'] == 'master':
                    LOG.debug("Found 'master' instance, calling grow on guest")
                    master_instance = Instance.load(context,
                                                    db_instance.id)
                    self.get_guest(master_instance).grow_cluster(new_ips)
                    break

            for guest in new_guests:
                guest.cluster_complete()

        timeout = Timeout(CONF.cluster_usage_timeout)

        try:
            _grow_cluster()
            self.reset_task()
        except Timeout as t:
            if t is not timeout:
                raise  # not my timeout
            LOG.exception(_("Timeout for growing cluster."))
            self.update_statuses_on_failure(cluster_id)
        except Exception:
            LOG.exception(_("Error growing cluster %s.") % cluster_id)
            self.update_statuses_on_failure(cluster_id)
        finally:
            timeout.cancel()

def execute_with_timeout(*args, **kwargs):
    time = kwargs.pop('timeout', 30)
    log_output_on_error = kwargs.pop('log_output_on_error', False)

    timeout = Timeout(time)
    try:
        return execute(*args, **kwargs)
    except exception.ProcessExecutionError as e:
        if log_output_on_error:
            LOG.error(
                _("Command '%(cmd)s' failed. %(description)s "
                  "Exit code: %(exit_code)s\nstderr: %(stderr)s\n"
                  "stdout: %(stdout)s") %
                {'cmd': e.cmd, 'description': e.description or '',
                 'exit_code': e.exit_code, 'stderr': e.stderr,
                 'stdout': e.stdout})
        raise
    except Timeout as t:
        if t is not timeout:
            LOG.error(_("Got a timeout but not the one expected."))
            raise
        else:
            msg = (_("Time out after waiting "
                     "%(time)s seconds when running proc: %(args)s"
                     " %(kwargs)s.") % {'time': time, 'args': args,
                                        'kwargs': kwargs})
            LOG.error(msg)
            raise exception.ProcessExecutionError(msg)
    finally:
        timeout.cancel()