openstack
/
trove
Code Issues Proposed changes

Merge "Fix RPC Versioning"

This commit is contained in:
Jenkins 2016-11-16 20:09:43 +00:00 committed by Gerrit Code Review
parent df37e118c0 18f23a7177
commit 0d514081c9
14 changed files with 501 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
trove/cmd/conductor.py
View File

@ -16,13 +16,13 @@ from oslo_concurrency import processutils
from oslo_service import service as openstack_service
from trove.cmd.common import with_initialize
from trove.conductor import api as conductor_api
@with_initialize
def main(conf):
from trove.common import notification
from trove.common.rpc import service as rpc_service
from trove.common.rpc import version as rpc_version
from trove.instance import models as inst_models
notification.DBaaSAPINotification.register_notify_callback(
@ -30,7 +30,7 @@ def main(conf):
topic = conf.conductor_queue
server = rpc_service.RpcService(
manager=conf.conductor_manager, topic=topic,
rpc_api_version=rpc_version.RPC_API_VERSION)
rpc_api_version=conductor_api.API.API_LATEST_VERSION)
workers = conf.trove_conductor_workers or processutils.get_worker_count()
launcher = openstack_service.launch(conf, server, workers=workers)
launcher.wait()

4
trove/cmd/guest.py
View File

@ -25,6 +25,7 @@ from oslo_service import service as openstack_service
from trove.common import cfg
from trove.common import debug_utils
from trove.common.i18n import _LE
from trove.guestagent import api as guest_api
CONF = cfg.CONF
# The guest_id opt definition must match the one in common/cfg.py
@ -57,11 +58,10 @@ def main():
rpc.init(CONF)
from trove.common.rpc import service as rpc_service
from trove.common.rpc import version as rpc_version
server = rpc_service.RpcService(
topic="guestagent.%s" % CONF.guest_id,
manager=manager, host=CONF.guest_id,
rpc_api_version=rpc_version.RPC_API_VERSION)
rpc_api_version=guest_api.API.API_LATEST_VERSION)
launcher = openstack_service.launch(CONF, server)
launcher.wait()

4
trove/cmd/taskmanager.py
View File

@ -16,6 +16,7 @@ from oslo_config import cfg as openstack_cfg
from oslo_service import service as openstack_service
from trove.cmd.common import with_initialize
from trove.taskmanager import api as task_api
extra_opts = [openstack_cfg.StrOpt('taskmanager_manager')]
@ -24,14 +25,13 @@ extra_opts = [openstack_cfg.StrOpt('taskmanager_manager')]
def startup(conf, topic):
from trove.common import notification
from trove.common.rpc import service as rpc_service
from trove.common.rpc import version as rpc_version
from trove.instance import models as inst_models
notification.DBaaSAPINotification.register_notify_callback(
inst_models.persist_instance_fault)
server = rpc_service.RpcService(
manager=conf.taskmanager_manager, topic=topic,
rpc_api_version=rpc_version.RPC_API_VERSION)
rpc_api_version=task_api.API.API_LATEST_VERSION)
launcher = openstack_service.launch(conf, server)
launcher.wait()

11
trove/common/cfg.py
View File

@ -1449,18 +1449,19 @@ mariadb_opts = [
upgrade_levels = cfg.OptGroup(
'upgrade_levels',
title='RPC upgrade levels group for handling versions',
help='Contains the support version caps for each RPC API')
help='Contains the support version caps (Openstack Release) for '
'each RPC API')
rpcapi_cap_opts = [
cfg.StrOpt(
'taskmanager', default="icehouse",
'taskmanager', default='latest',
help='Set a version cap for messages sent to taskmanager services'),
cfg.StrOpt(
'guestagent', default="icehouse",
'guestagent', default='latest',
help='Set a version cap for messages sent to guestagent services'),
cfg.StrOpt(
'conductor', default="icehouse",
help='Set a version cap for messages sent to conductor services'),
'conductor', default='latest',
help='Set Openstack Release compatibility for conductor services'),
]
CONF = cfg.CONF

59
trove/common/strategies/cluster/experimental/cassandra/guestagent.py
View File

@ -30,65 +30,100 @@ class CassandraGuestAgentStrategy(base.BaseGuestAgentStrategy):
class CassandraGuestAgentAPI(guest_api.API):
"""Cluster Specific Datastore Guest API
**** VERSION CONTROLLED API ****
The methods in this class are subject to version control as
coordinated by guestagent/api.py. Whenever a change is made to
any API method in this class, add a version number and comment
to the top of guestagent/api.py and use the version number as
appropriate in this file
"""
def get_data_center(self):
LOG.debug("Retrieving the data center for node: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("get_data_center", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def get_rack(self):
LOG.debug("Retrieving the rack for node: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("get_rack", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def set_seeds(self, seeds):
LOG.debug("Configuring the gossip seeds for node: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("set_seeds", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap, seeds=seeds)
version=version, seeds=seeds)
def get_seeds(self):
LOG.debug("Retrieving the gossip seeds for node: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("get_seeds", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def set_auto_bootstrap(self, enabled):
LOG.debug("Setting the auto-bootstrap to '%s' for node: %s"
% (enabled, self.id))
version = guest_api.API.API_BASE_VERSION
return self._call("set_auto_bootstrap", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap, enabled=enabled)
version=version, enabled=enabled)
def cluster_complete(self):
LOG.debug("Sending a setup completion notification for node: %s"
% self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("cluster_complete", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
def node_cleanup_begin(self):
LOG.debug("Signaling the node to prepare for cleanup: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("node_cleanup_begin", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def node_cleanup(self):
LOG.debug("Running cleanup on node: %s" % self.id)
return self._cast('node_cleanup', self.version_cap)
version = guest_api.API.API_BASE_VERSION
return self._cast('node_cleanup', version=version)
def node_decommission(self):
LOG.debug("Decommission node: %s" % self.id)
return self._cast("node_decommission", self.version_cap)
version = guest_api.API.API_BASE_VERSION
return self._cast("node_decommission", version=version)
def cluster_secure(self, password):
LOG.debug("Securing the cluster via node: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call(
"cluster_secure", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap, password=password)
version=version, password=password)
def get_admin_credentials(self):
LOG.debug("Retrieving the admin credentials from node: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("get_admin_credentials", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def store_admin_credentials(self, admin_credentials):
LOG.debug("Storing the admin credentials on node: %s" % self.id)
version = guest_api.API.API_BASE_VERSION
return self._call("store_admin_credentials",
guest_api.AGENT_LOW_TIMEOUT, self.version_cap,
guest_api.AGENT_LOW_TIMEOUT,
version=version,
admin_credentials=admin_credentials)

30
trove/common/strategies/cluster/experimental/galera_common/guestagent.py
View File

@ -31,39 +31,59 @@ class GaleraCommonGuestAgentStrategy(cluster_base.BaseGuestAgentStrategy):
class GaleraCommonGuestAgentAPI(guest_api.API):
"""Cluster Specific Datastore Guest API
**** VERSION CONTROLLED API ****
The methods in this class are subject to version control as
coordinated by guestagent/api.py. Whenever a change is made to
any API method in this class, add a version number and comment
to the top of guestagent/api.py and use the version number as
appropriate in this file
"""
def install_cluster(self, replication_user, cluster_configuration,
bootstrap):
"""Install the cluster."""
LOG.debug("Installing Galera cluster.")
version = guest_api.API.API_BASE_VERSION
self._call("install_cluster", CONF.cluster_usage_timeout,
self.version_cap,
version=version,
replication_user=replication_user,
cluster_configuration=cluster_configuration,
bootstrap=bootstrap)
def reset_admin_password(self, admin_password):
"""Store this password on the instance as the admin password."""
version = guest_api.API.API_BASE_VERSION
self._call("reset_admin_password", CONF.cluster_usage_timeout,
self.version_cap,
version=version,
admin_password=admin_password)
def cluster_complete(self):
"""Set the status that the cluster is build is complete."""
LOG.debug("Notifying cluster install completion.")
version = guest_api.API.API_BASE_VERSION
return self._call("cluster_complete", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
def get_cluster_context(self):
"""Get the context of the cluster."""
LOG.debug("Getting the cluster context.")
version = guest_api.API.API_BASE_VERSION
return self._call("get_cluster_context", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
def write_cluster_configuration_overrides(self, cluster_configuration):
"""Write an updated the cluster configuration."""
LOG.debug("Writing an updated the cluster configuration.")
version = guest_api.API.API_BASE_VERSION
self._call("write_cluster_configuration_overrides",
guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap,
version=version,
cluster_configuration=cluster_configuration)

55
trove/common/strategies/cluster/experimental/mongodb/guestagent.py
View File

@ -33,6 +33,16 @@ class MongoDbGuestAgentStrategy(base.BaseGuestAgentStrategy):
class MongoDbGuestAgentAPI(guest_api.API):
"""Cluster Specific Datastore Guest API
**** VERSION CONTROLLED API ****
The methods in this class are subject to version control as
coordinated by guestagent/api.py. Whenever a change is made to
any API method in this class, add a version number and comment
to the top of guestagent/api.py and use the version number as
appropriate in this file
"""
def add_shard(self, replica_set_name, replica_set_member):
LOG.debug("Adding shard with replSet %(replica_set_name)s and member "
@ -40,66 +50,89 @@ class MongoDbGuestAgentAPI(guest_api.API):
"%(id)s" % {'replica_set_name': replica_set_name,
'replica_set_member': replica_set_member,
'id': self.id})
version = guest_api.API.API_BASE_VERSION
return self._call("add_shard", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap,
version=version,
replica_set_name=replica_set_name,
replica_set_member=replica_set_member)
def add_members(self, members):
LOG.debug("Adding members %(members)s on instance %(id)s" % {
'members': members, 'id': self.id})
version = guest_api.API.API_BASE_VERSION
return self._call("add_members", ADD_MEMBERS_TIMEOUT,
self.version_cap, members=members)
version=version, members=members)
def add_config_servers(self, config_servers):
LOG.debug("Adding config servers %(config_servers)s for instance "
"%(id)s" % {'config_servers': config_servers,
'id': self.id})
version = guest_api.API.API_BASE_VERSION
return self._call("add_config_servers", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap, config_servers=config_servers)
version=version,
config_servers=config_servers)
def cluster_complete(self):
LOG.debug("Notify regarding cluster install completion")
version = guest_api.API.API_BASE_VERSION
return self._call("cluster_complete", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
def get_key(self):
LOG.debug("Requesting cluster key from guest")
version = guest_api.API.API_BASE_VERSION
return self._call("get_key", guest_api.AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def prep_primary(self):
LOG.debug("Preparing member to be primary member.")
version = guest_api.API.API_BASE_VERSION
return self._call("prep_primary", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
def create_admin_user(self, password):
LOG.debug("Creating admin user")
version = guest_api.API.API_BASE_VERSION
return self._call("create_admin_user", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap, password=password)
version=version, password=password)
def store_admin_password(self, password):
LOG.debug("Storing admin password")
version = guest_api.API.API_BASE_VERSION
return self._call("store_admin_password",
guest_api.AGENT_LOW_TIMEOUT,
self.version_cap,
version=version,
password=password)
def get_replica_set_name(self):
LOG.debug("Querying member for its replica set name")
version = guest_api.API.API_BASE_VERSION
return self._call("get_replica_set_name",
guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
def get_admin_password(self):
LOG.debug("Querying instance for its admin password")
version = guest_api.API.API_BASE_VERSION
return self._call("get_admin_password",
guest_api.AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def is_shard_active(self, replica_set_name):
LOG.debug("Checking if replica set %s is active" % replica_set_name)
version = guest_api.API.API_BASE_VERSION
return self._call("is_shard_active",
guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap,
version=version,
replica_set_name=replica_set_name)

3
trove/common/strategies/cluster/experimental/mongodb/taskmanager.py
View File

@ -379,7 +379,8 @@ class MongoDbTaskManagerAPI(task_api.API):
def mongodb_add_shard_cluster(self, cluster_id, shard_id,
replica_set_name):
LOG.debug("Making async call to add shard cluster %s " % cluster_id)
cctxt = self.client.prepare(version=self.version_cap)
version = task_api.API.API_BASE_VERSION
cctxt = self.client.prepare(version=version)
cctxt.cast(self.context,
"add_shard_cluster",
cluster_id=cluster_id,

37
trove/common/strategies/cluster/experimental/redis/guestagent.py
View File

@ -28,34 +28,59 @@ class RedisGuestAgentStrategy(base.BaseGuestAgentStrategy):
class RedisGuestAgentAPI(guest_api.API):
"""Cluster Specific Datastore Guest API
**** VERSION CONTROLLED API ****
The methods in this class are subject to version control as
coordinated by guestagent/api.py. Whenever a change is made to
any API method in this class, add a version number and comment
to the top of guestagent/api.py and use the version number as
appropriate in this file
"""
def get_node_ip(self):
LOG.debug("Retrieve ip info from node.")
version = guest_api.API.API_BASE_VERSION
return self._call("get_node_ip",
guest_api.AGENT_HIGH_TIMEOUT, self.version_cap)
guest_api.AGENT_HIGH_TIMEOUT,
version=version)
def get_node_id_for_removal(self):
LOG.debug("Validating cluster node removal.")
version = guest_api.API.API_BASE_VERSION
return self._call("get_node_id_for_removal",
guest_api.AGENT_HIGH_TIMEOUT, self.version_cap)
guest_api.AGENT_HIGH_TIMEOUT,
version=version)
def remove_nodes(self, node_ids):
LOG.debug("Removing nodes from cluster.")
version = guest_api.API.API_BASE_VERSION
return self._call("remove_nodes", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap, node_ids=node_ids)
version=version, node_ids=node_ids)
def cluster_meet(self, ip, port):
LOG.debug("Joining node to cluster.")
version = guest_api.API.API_BASE_VERSION
return self._call("cluster_meet", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap, ip=ip, port=port)
version=version, ip=ip, port=port)
def cluster_addslots(self, first_slot, last_slot):
LOG.debug("Adding slots %s-%s to cluster.", first_slot, last_slot)
version = guest_api.API.API_BASE_VERSION
return self._call("cluster_addslots",
guest_api.AGENT_HIGH_TIMEOUT, self.version_cap,
guest_api.AGENT_HIGH_TIMEOUT,
version=version,
first_slot=first_slot, last_slot=last_slot)
def cluster_complete(self):
LOG.debug("Notifying cluster install completion.")
version = guest_api.API.API_BASE_VERSION
return self._call("cluster_complete", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)

39
trove/common/strategies/cluster/experimental/vertica/guestagent.py
View File

@ -30,39 +30,64 @@ class VerticaGuestAgentStrategy(base.BaseGuestAgentStrategy):
class VerticaGuestAgentAPI(guest_api.API):
"""Cluster Specific Datastore Guest API
**** VERSION CONTROLLED API ****
The methods in this class are subject to version control as
coordinated by guestagent/api.py. Whenever a change is made to
any API method in this class, add a version number and comment
to the top of guestagent/api.py and use the version number as
appropriate in this file
"""
def get_public_keys(self, user):
LOG.debug("Getting public keys for user: %s." % user)
version = guest_api.API.API_BASE_VERSION
return self._call("get_public_keys", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap, user=user)
version=version, user=user)
def authorize_public_keys(self, user, public_keys):
LOG.debug("Authorizing public keys for user: %s." % user)
version = guest_api.API.API_BASE_VERSION
return self._call("authorize_public_keys",
guest_api.AGENT_HIGH_TIMEOUT, self.version_cap,
guest_api.AGENT_HIGH_TIMEOUT,
version=version,
user=user, public_keys=public_keys)
def install_cluster(self, members):
LOG.debug("Installing Vertica cluster on members: %s." % members)
version = guest_api.API.API_BASE_VERSION
return self._call("install_cluster", CONF.cluster_usage_timeout,
self.version_cap, members=members)
version=version, members=members)
def grow_cluster(self, members):
LOG.debug("Growing Vertica cluster with members: %s." % members)
version = guest_api.API.API_BASE_VERSION
return self._call("grow_cluster", CONF.cluster_usage_timeout,
self.version_cap, members=members)
version=version, members=members)
def shrink_cluster(self, members):
LOG.debug("Shrinking Vertica cluster with members: %s." % members)
version = guest_api.API.API_BASE_VERSION
return self._call("shrink_cluster", CONF.cluster_usage_timeout,
self.version_cap, members=members)
version=version, members=members)
def mark_design_ksafe(self, k):
LOG.debug("Setting vertica k-safety level to : %s." % k)
version = guest_api.API.API_BASE_VERSION
return self._call("mark_design_ksafe", CONF.cluster_usage_timeout,
self.version_cap, k=k)
version=version, k=k)
def cluster_complete(self):
LOG.debug("Notifying cluster install completion.")
version = guest_api.API.API_BASE_VERSION
return self._call("cluster_complete", guest_api.AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)

51
trove/conductor/api.py
View File

@ -16,28 +16,50 @@ from oslo_log import log as logging
import oslo_messaging as messaging
from trove.common import cfg
from trove.common.rpc import version as rpc_version
from trove.common.serializable_notification import SerializableNotification
from trove import rpc
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
class API(object):
"""API for interacting with trove conductor."""
"""API for interacting with trove conductor.
API version history:
* 1.0 - Initial version.
When updating this API, also update API_LATEST_VERSION
"""
# API_LATEST_VERSION should bump the minor number each time
# a method signature is added or changed
API_LATEST_VERSION = '1.0'
# API_BASE_VERSION should only change on major version upgrade
API_BASE_VERSION = '1.0'
VERSION_ALIASES = {
'icehouse': '1.0',
'juno': '1.0',
'kilo': '1.0',
'liberty': '1.0',
'mitaka': '1.0',
'newton': '1.0',
'latest': API_LATEST_VERSION
}
def __init__(self, context):
self.context = context
super(API, self).__init__()
version_cap = self.VERSION_ALIASES.get(
CONF.upgrade_levels.conductor, CONF.upgrade_levels.conductor)
target = messaging.Target(topic=CONF.conductor_queue,
version=rpc_version.RPC_API_VERSION)
version=version_cap)
self.version_cap = rpc_version.VERSION_ALIASES.get(
CONF.upgrade_levels.conductor)
self.client = self.get_client(target, self.version_cap)
self.client = self.get_client(target, version_cap)
def get_client(self, target, version_cap, serializer=None):
return rpc.get_client(target,
@ -47,8 +69,9 @@ class API(object):
def heartbeat(self, instance_id, payload, sent=None):
LOG.debug("Making async call to cast heartbeat for instance: %s"
% instance_id)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=self.version_cap)
cctxt = self.client.prepare(version=version)
cctxt.cast(self.context, "heartbeat",
instance_id=instance_id,
sent=sent,
@ -58,8 +81,9 @@ class API(object):
**backup_fields):
LOG.debug("Making async call to cast update_backup for instance: %s"
% instance_id)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=self.version_cap)
cctxt = self.client.prepare(version=version)
cctxt.cast(self.context, "update_backup",
instance_id=instance_id,
backup_id=backup_id,
@ -69,14 +93,16 @@ class API(object):
def report_root(self, instance_id, user):
LOG.debug("Making async call to cast report_root for instance: %s"
% instance_id)
cctxt = self.client.prepare(version=self.version_cap)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=version)
cctxt.cast(self.context, "report_root",
instance_id=instance_id,
user=user)
def notify_end(self, **notification_args):
LOG.debug("Making async call to cast end notification")
cctxt = self.client.prepare(version=self.version_cap)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=version)
context = self.context
serialized = SerializableNotification.serialize(context,
context.notification)
@ -86,7 +112,8 @@ class API(object):
def notify_exc_info(self, message, exception):
LOG.debug("Making async call to cast error notification")
cctxt = self.client.prepare(version=self.version_cap)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=version)
context = self.context
serialized = SerializableNotification.serialize(context,
context.notification)

268
trove/guestagent/api.py
View File

@ -26,7 +26,6 @@ from trove.common import cfg
from trove.common import exception
from trove.common.i18n import _
from trove.common.notification import NotificationCastWrapper
import trove.common.rpc.version as rpc_version
from trove import rpc
CONF = cfg.CONF
@ -37,19 +36,43 @@ AGENT_SNAPSHOT_TIMEOUT = CONF.agent_replication_snapshot_timeout
class API(object):
"""API for interacting with the guest manager."""
"""API for interacting with the guest manager.
API version history:
* 1.0 - Initial version.
When updating this API, also update API_LATEST_VERSION
"""
# API_LATEST_VERSION should bump the minor number each time
# a method signature is added or changed
API_LATEST_VERSION = '1.0'
# API_BASE_VERSION should only change on major version upgrade
API_BASE_VERSION = '1.0'
VERSION_ALIASES = {
'icehouse': '1.0',
'juno': '1.0',
'kilo': '1.0',
'liberty': '1.0',
'mitaka': '1.0',
'newton': '1.0',
'latest': API_LATEST_VERSION
}
def __init__(self, context, id):
self.context = context
self.id = id
super(API, self).__init__()
version_cap = self.VERSION_ALIASES.get(
CONF.upgrade_levels.guestagent, CONF.upgrade_levels.guestagent)
target = messaging.Target(topic=self._get_routing_key(),
version=rpc_version.RPC_API_VERSION)
version=version_cap)
self.version_cap = rpc_version.VERSION_ALIASES.get(
CONF.upgrade_levels.guestagent)
self.client = self.get_client(target, self.version_cap)
self.client = self.get_client(target, version_cap)
def get_client(self, target, version_cap, serializer=None):
return rpc.get_client(target,
@ -95,31 +118,44 @@ class API(object):
users.
"""
LOG.debug("Changing passwords for users on instance %s.", self.id)
self._cast("change_passwords", self.version_cap, users=users)
version = self.API_BASE_VERSION
self._cast("change_passwords", version=version, users=users)
def update_attributes(self, username, hostname, user_attrs):
"""Update user attributes."""
LOG.debug("Changing user attributes on instance %s.", self.id)
self._cast("update_attributes", self.version_cap, username=username,
version = self.API_BASE_VERSION
self._cast("update_attributes",
version=version, username=username,
hostname=hostname, user_attrs=user_attrs)
def create_user(self, users):
"""Make an asynchronous call to create a new database user"""
LOG.debug("Creating Users for instance %s.", self.id)
self._cast("create_user", self.version_cap, users=users)
version = self.API_BASE_VERSION
self._cast("create_user", version=version, users=users)
def get_user(self, username, hostname):
"""Make an asynchronous call to get a single database user."""
LOG.debug("Getting a user %(username)s on instance %(id)s.",
{'username': username, 'id': self.id})
return self._call("get_user", AGENT_LOW_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
return self._call("get_user",
AGENT_LOW_TIMEOUT, version=version,
username=username, hostname=hostname)
def list_access(self, username, hostname):
"""Show all the databases to which a user has more than USAGE."""
LOG.debug("Showing user %(username)s grants on instance %(id)s.",
{'username': username, 'id': self.id})
return self._call("list_access", AGENT_LOW_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
return self._call("list_access",
AGENT_LOW_TIMEOUT, version=version,
username=username, hostname=hostname)
def grant_access(self, username, hostname, databases):
@ -128,7 +164,10 @@ class API(object):
"%(username)s on instance %(id)s.", {'username': username,
'databases': databases,
'id': self.id})
return self._call("grant_access", AGENT_LOW_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
return self._call("grant_access",
AGENT_LOW_TIMEOUT, version=version,
username=username, hostname=hostname,
databases=databases)
@ -138,14 +177,20 @@ class API(object):
"%(username)s on instance %(id)s.", {'username': username,
'database': database,
'id': self.id})
return self._call("revoke_access", AGENT_LOW_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
return self._call("revoke_access",
AGENT_LOW_TIMEOUT, version=version,
username=username, hostname=hostname,
database=database)
def list_users(self, limit=None, marker=None, include_marker=False):
"""Make an asynchronous call to list database users."""
LOG.debug("Listing Users for instance %s.", self.id)
return self._call("list_users", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
return self._call("list_users", AGENT_HIGH_TIMEOUT,
version=version,
limit=limit, marker=marker,
include_marker=include_marker)
@ -153,20 +198,27 @@ class API(object):
"""Make an asynchronous call to delete an existing database user."""
LOG.debug("Deleting user %(user)s for instance %(instance_id)s." %
{'user': user, 'instance_id': self.id})
self._cast("delete_user", self.version_cap, user=user)
version = self.API_BASE_VERSION
self._cast("delete_user", version=version, user=user)
def create_database(self, databases):
"""Make an asynchronous call to create a new database
within the specified container
"""
LOG.debug("Creating databases for instance %s.", self.id)
self._cast("create_database", self.version_cap, databases=databases)
version = self.API_BASE_VERSION
self._cast("create_database", version=version,
databases=databases)
def list_databases(self, limit=None, marker=None, include_marker=False):
"""Make an asynchronous call to list databases."""
LOG.debug("Listing databases for instance %s.", self.id)
version = self.API_BASE_VERSION
return self._call("list_databases", AGENT_LOW_TIMEOUT,
self.version_cap, limit=limit, marker=marker,
version=version, limit=limit, marker=marker,
include_marker=include_marker)
def delete_database(self, database):
@ -176,53 +228,72 @@ class API(object):
LOG.debug("Deleting database %(database)s for "
"instance %(instance_id)s." % {'database': database,
'instance_id': self.id})
self._cast("delete_database", self.version_cap, database=database)
version = self.API_BASE_VERSION
self._cast("delete_database", version=version, database=database)
def enable_root(self):
"""Make a synchronous call to enable the root user for
access from anywhere
"""
LOG.debug("Enable root user for instance %s.", self.id)
return self._call("enable_root", AGENT_HIGH_TIMEOUT, self.version_cap)
version = self.API_BASE_VERSION
return self._call("enable_root", AGENT_HIGH_TIMEOUT,
version=version)
def enable_root_with_password(self, root_password=None):
"""Make a synchronous call to enable the root user for
access from anywhere
"""
LOG.debug("Enable root user for instance %s.", self.id)
version = self.API_BASE_VERSION
return self._call("enable_root_with_password", AGENT_HIGH_TIMEOUT,
self.version_cap, root_password=root_password)
version=version, root_password=root_password)
def disable_root(self):
"""Make a synchronous call to disable the root user for
access from anywhere
"""
LOG.debug("Disable root user for instance %s.", self.id)
return self._call("disable_root", AGENT_LOW_TIMEOUT, self.version_cap)
version = self.API_BASE_VERSION
return self._call("disable_root", AGENT_LOW_TIMEOUT,
version=version)
def is_root_enabled(self):
"""Make a synchronous call to check if root access is
available for the container
"""
LOG.debug("Check root access for instance %s.", self.id)
version = self.API_BASE_VERSION
return self._call("is_root_enabled", AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def get_hwinfo(self):
"""Make a synchronous call to get hardware info for the container"""
LOG.debug("Check hwinfo on instance %s.", self.id)
return self._call("get_hwinfo", AGENT_LOW_TIMEOUT, self.version_cap)
version = self.API_BASE_VERSION
return self._call("get_hwinfo", AGENT_LOW_TIMEOUT,
version=version)
def get_diagnostics(self):
"""Make a synchronous call to get diagnostics for the container"""
LOG.debug("Check diagnostics on instance %s.", self.id)
version = self.API_BASE_VERSION
return self._call("get_diagnostics", AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def rpc_ping(self):
"""Make a synchronous RPC call to check if we can ping the instance."""
LOG.debug("Check RPC ping on instance %s.", self.id)
return self._call("rpc_ping", AGENT_LOW_TIMEOUT, self.version_cap)
version = self.API_BASE_VERSION
return self._call("rpc_ping", AGENT_LOW_TIMEOUT, version=version)
def prepare(self, memory_mb, packages, databases, users,
device_path='/dev/vdb', mount_point='/mnt/volume',
@ -234,6 +305,8 @@ class API(object):
"""
LOG.debug("Sending the call to prepare the Guest.")
version = self.API_BASE_VERSION
# Taskmanager is a publisher, guestagent is a consumer. Usually
# consumer creates a queue, but in this case we have to make sure
# "prepare" doesn't get lost if for some reason guest was delayed and
@ -242,7 +315,7 @@ class API(object):
packages = packages.split()
self._cast(
"prepare", self.version_cap, packages=packages,
"prepare", version=version, packages=packages,
databases=databases, memory_mb=memory_mb, users=users,
device_path=device_path, mount_point=mount_point,
backup_info=backup_info, config_contents=config_contents,
@ -258,7 +331,7 @@ class API(object):
server = None
target = messaging.Target(topic=self._get_routing_key(),
server=self.id,
version=rpc_version.RPC_API_VERSION)
version=self.API_BASE_VERSION)
try:
server = rpc.get_server(target, [])
server.start()
@ -270,26 +343,35 @@ class API(object):
def pre_upgrade(self):
"""Prepare the guest for upgrade."""
LOG.debug("Sending the call to prepare the guest for upgrade.")
return self._call("pre_upgrade", AGENT_HIGH_TIMEOUT, self.version_cap)
version = self.API_BASE_VERSION
return self._call("pre_upgrade", AGENT_HIGH_TIMEOUT,
version=version)
def post_upgrade(self, upgrade_info):
"""Recover the guest after upgrading the guest's image."""
LOG.debug("Recover the guest after upgrading the guest's image.")
self._call("post_upgrade", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("post_upgrade", AGENT_HIGH_TIMEOUT, version=version,
upgrade_info=upgrade_info)
def restart(self):
"""Restart the database server."""
LOG.debug("Sending the call to restart the database process "
"on the Guest.")
self._call("restart", AGENT_HIGH_TIMEOUT, self.version_cap)
version = self.API_BASE_VERSION
self._call("restart", AGENT_HIGH_TIMEOUT, version=version)
def start_db_with_conf_changes(self, config_contents):
"""Start the database server."""
LOG.debug("Sending the call to start the database process on "
"the Guest with a timeout of %s." % AGENT_HIGH_TIMEOUT)
version = self.API_BASE_VERSION
self._call("start_db_with_conf_changes", AGENT_HIGH_TIMEOUT,
self.version_cap, config_contents=config_contents)
version=version, config_contents=config_contents)
def reset_configuration(self, configuration):
"""Ignore running state of the database server; just change
@ -297,20 +379,26 @@ class API(object):
"""
LOG.debug("Sending the call to change the database conf file on the "
"Guest with a timeout of %s." % AGENT_HIGH_TIMEOUT)
version = self.API_BASE_VERSION
self._call("reset_configuration", AGENT_HIGH_TIMEOUT,
self.version_cap, configuration=configuration)
version=version, configuration=configuration)
def stop_db(self, do_not_start_on_reboot=False):
"""Stop the database server."""
LOG.debug("Sending the call to stop the database process "
"on the Guest.")
self._call("stop_db", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("stop_db", AGENT_HIGH_TIMEOUT, version=version,
do_not_start_on_reboot=do_not_start_on_reboot)
def upgrade(self, instance_version, location, metadata=None):
"""Make an asynchronous call to self upgrade the guest agent."""
LOG.debug("Sending an upgrade call to nova-guest.")
self._cast("upgrade", self.version_cap,
version = self.API_BASE_VERSION
self._cast("upgrade", version=version,
instance_version=instance_version,
location=location,
metadata=metadata)
@ -318,158 +406,214 @@ class API(object):
def get_volume_info(self):
"""Make a synchronous call to get volume info for the container."""
LOG.debug("Check Volume Info on instance %s.", self.id)
version = self.API_BASE_VERSION
return self._call("get_filesystem_stats", AGENT_LOW_TIMEOUT,
self.version_cap, fs_path=None)
version=version, fs_path=None)
def update_guest(self):
"""Make a synchronous call to update the guest agent."""
LOG.debug("Updating guest agent on instance %s.", self.id)
self._call("update_guest", AGENT_HIGH_TIMEOUT, self.version_cap)
version = self.API_BASE_VERSION
self._call("update_guest", AGENT_HIGH_TIMEOUT, version=version)
def create_backup(self, backup_info):
"""Make async call to create a full backup of this instance."""
LOG.debug("Create Backup %(backup_id)s "
"for instance %(instance_id)s." %
{'backup_id': backup_info['id'], 'instance_id': self.id})
self._cast("create_backup", self.version_cap, backup_info=backup_info)
version = self.API_BASE_VERSION
self._cast("create_backup", version=version,
backup_info=backup_info)
def mount_volume(self, device_path=None, mount_point=None):
"""Mount the volume."""
LOG.debug("Mount volume %(mount)s on instance %(id)s." % {
'mount': mount_point, 'id': self.id})
self._call("mount_volume", AGENT_LOW_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("mount_volume", AGENT_LOW_TIMEOUT, version=version,
device_path=device_path, mount_point=mount_point)
def unmount_volume(self, device_path=None, mount_point=None):
"""Unmount the volume."""
LOG.debug("Unmount volume %(device)s on instance %(id)s." % {
'device': device_path, 'id': self.id})
self._call("unmount_volume", AGENT_LOW_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("unmount_volume", AGENT_LOW_TIMEOUT, version=version,
device_path=device_path, mount_point=mount_point)
def resize_fs(self, device_path=None, mount_point=None):
"""Resize the filesystem."""
LOG.debug("Resize device %(device)s on instance %(id)s." % {
'device': device_path, 'id': self.id})
self._call("resize_fs", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("resize_fs", AGENT_HIGH_TIMEOUT, version=version,
device_path=device_path, mount_point=mount_point)
def update_overrides(self, overrides, remove=False):
"""Update the overrides."""
LOG.debug("Updating overrides values %(overrides)s on instance "
"%(id)s.", {'overrides': overrides, 'id': self.id})
version = self.API_BASE_VERSION
self._call("update_overrides", AGENT_HIGH_TIMEOUT,
self.version_cap, overrides=overrides, remove=remove)
version=version, overrides=overrides, remove=remove)
def apply_overrides(self, overrides):
LOG.debug("Applying overrides values %(overrides)s on instance "
"%(id)s.", {'overrides': overrides, 'id': self.id})
self._call("apply_overrides", AGENT_HIGH_TIMEOUT, self.version_cap,
overrides=overrides)
version = self.API_BASE_VERSION
self._call("apply_overrides", AGENT_HIGH_TIMEOUT,
version=version, overrides=overrides)
def backup_required_for_replication(self):
LOG.debug("Checking backup requirement for replication")
version = self.API_BASE_VERSION
return self._call("backup_required_for_replication",
AGENT_LOW_TIMEOUT,
self.version_cap)
version=version)
def get_replication_snapshot(self, snapshot_info=None,
replica_source_config=None):
LOG.debug("Retrieving replication snapshot from instance %s.", self.id)
version = self.API_BASE_VERSION
return self._call("get_replication_snapshot", AGENT_SNAPSHOT_TIMEOUT,
self.version_cap, snapshot_info=snapshot_info,
version=version, snapshot_info=snapshot_info,
replica_source_config=replica_source_config)
def attach_replication_slave(self, snapshot, replica_config=None):
LOG.debug("Configuring instance %s to replicate from %s.",
self.id, snapshot.get('master').get('id'))
self._cast("attach_replication_slave", self.version_cap,
version = self.API_BASE_VERSION
self._cast("attach_replication_slave", version=version,
snapshot=snapshot, slave_config=replica_config)
def detach_replica(self, for_failover=False):
LOG.debug("Detaching replica %s from its replication source.", self.id)
version = self.API_BASE_VERSION
return self._call("detach_replica", AGENT_HIGH_TIMEOUT,
self.version_cap, for_failover=for_failover)
version=version, for_failover=for_failover)
def get_replica_context(self):
LOG.debug("Getting replica context.")
version = self.API_BASE_VERSION
return self._call("get_replica_context",
AGENT_HIGH_TIMEOUT, self.version_cap)
AGENT_HIGH_TIMEOUT, version=version)
def attach_replica(self, replica_info, slave_config):
LOG.debug("Attaching replica %s." % replica_info)
self._call("attach_replica", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("attach_replica", AGENT_HIGH_TIMEOUT, version=version,
replica_info=replica_info, slave_config=slave_config)
def make_read_only(self, read_only):
LOG.debug("Executing make_read_only(%s)" % read_only)
self._call("make_read_only", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("make_read_only", AGENT_HIGH_TIMEOUT, version=version,
read_only=read_only)
def enable_as_master(self, replica_source_config):
LOG.debug("Executing enable_as_master")
self._call("enable_as_master", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("enable_as_master", AGENT_HIGH_TIMEOUT,
version=version,
replica_source_config=replica_source_config)
# DEPRECATED: Maintain for API Compatibility
def get_txn_count(self):
LOG.debug("Executing get_txn_count.")
version = self.API_BASE_VERSION
return self._call("get_txn_count",
AGENT_HIGH_TIMEOUT, self.version_cap)
AGENT_HIGH_TIMEOUT, version=version)
def get_last_txn(self):
LOG.debug("Executing get_last_txn.")
version = self.API_BASE_VERSION
return self._call("get_last_txn",
AGENT_HIGH_TIMEOUT, self.version_cap)
AGENT_HIGH_TIMEOUT, version=version)
def get_latest_txn_id(self):
LOG.debug("Executing get_latest_txn_id.")
version = self.API_BASE_VERSION
return self._call("get_latest_txn_id",
AGENT_HIGH_TIMEOUT, self.version_cap)
AGENT_HIGH_TIMEOUT, version=version)
def wait_for_txn(self, txn):
LOG.debug("Executing wait_for_txn.")
self._call("wait_for_txn", AGENT_HIGH_TIMEOUT, self.version_cap,
version = self.API_BASE_VERSION
self._call("wait_for_txn", AGENT_HIGH_TIMEOUT, version=version,
txn=txn)
def cleanup_source_on_replica_detach(self, replica_info):
LOG.debug("Cleaning up master %s on detach of replica.", self.id)
version = self.API_BASE_VERSION
self._call("cleanup_source_on_replica_detach", AGENT_HIGH_TIMEOUT,
self.version_cap, replica_info=replica_info)
version=version, replica_info=replica_info)
def demote_replication_master(self):
LOG.debug("Demoting instance %s to non-master.", self.id)
version = self.API_BASE_VERSION
self._call("demote_replication_master", AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
def guest_log_list(self):
LOG.debug("Retrieving guest log list for %s.", self.id)
version = self.API_BASE_VERSION
result = self._call("guest_log_list", AGENT_HIGH_TIMEOUT,
self.version_cap)
version=version)
LOG.debug("guest_log_list returns %s", result)
return result
def guest_log_action(self, log_name, enable, disable, publish, discard):
LOG.debug("Processing guest log '%s' for %s.", log_name, self.id)
version = self.API_BASE_VERSION
return self._call("guest_log_action", AGENT_HIGH_TIMEOUT,
self.version_cap, log_name=log_name,
version=version, log_name=log_name,
enable=enable, disable=disable,
publish=publish, discard=discard)
def module_list(self, include_contents):
LOG.debug("Querying modules on %s (contents: %s).",
self.id, include_contents)
version = self.API_BASE_VERSION
result = self._call("module_list", AGENT_HIGH_TIMEOUT,
self.version_cap,
version=version,
include_contents=include_contents)
return result
def module_apply(self, modules):
LOG.debug("Applying modules to %s.", self.id)
version = self.API_BASE_VERSION
return self._call("module_apply", AGENT_HIGH_TIMEOUT,
self.version_cap, modules=modules)
version=version, modules=modules)
def module_remove(self, module):
LOG.debug("Removing modules from %s.", self.id)
version = self.API_BASE_VERSION
return self._call("module_remove", AGENT_HIGH_TIMEOUT,
self.version_cap, module=module)
version=version, module=module)

86
trove/taskmanager/api.py
View File

@ -24,7 +24,6 @@ import oslo_messaging as messaging
from trove.common import cfg
from trove.common import exception
from trove.common.notification import NotificationCastWrapper
import trove.common.rpc.version as rpc_version
from trove.common.strategies.cluster import strategy
from trove.guestagent import models as agent_models
from trove import rpc
@ -34,18 +33,42 @@ LOG = logging.getLogger(__name__)
class API(object):
"""API for interacting with the task manager."""
"""API for interacting with the task manager.
API version history:
* 1.0 - Initial version.
When updating this API, also update API_LATEST_VERSION
"""
# API_LATEST_VERSION should bump the minor number each time
# a method signature is added or changed
API_LATEST_VERSION = '1.0'
# API_BASE_VERSION should only change on major version upgrade
API_BASE_VERSION = '1.0'
VERSION_ALIASES = {
'icehouse': '1.0',
'juno': '1.0',
'kilo': '1.0',
'liberty': '1.0',
'mitaka': '1.0',
'newton': '1.0',
'latest': API_LATEST_VERSION
}
def __init__(self, context):
self.context = context
super(API, self).__init__()
version_cap = self.VERSION_ALIASES.get(
CONF.upgrade_levels.taskmanager, CONF.upgrade_levels.taskmanager)
target = messaging.Target(topic=CONF.taskmanager_queue,
version=rpc_version.RPC_API_VERSION)
version=version_cap)
self.version_cap = rpc_version.VERSION_ALIASES.get(
CONF.upgrade_levels.taskmanager)
self.client = self.get_client(target, self.version_cap)
self.client = self.get_client(target, version_cap)
def _cast(self, method_name, version, **kwargs):
LOG.debug("Casting %s" % method_name)
@ -79,72 +102,83 @@ class API(object):
def resize_volume(self, new_size, instance_id):
LOG.debug("Making async call to resize volume for instance: %s"
% instance_id)
version = self.API_BASE_VERSION
self._cast("resize_volume", self.version_cap,
self._cast("resize_volume", version=version,
new_size=new_size,
instance_id=instance_id)
def resize_flavor(self, instance_id, old_flavor, new_flavor):
LOG.debug("Making async call to resize flavor for instance: %s" %
instance_id)
version = self.API_BASE_VERSION
self._cast("resize_flavor", self.version_cap,
self._cast("resize_flavor", version=version,
instance_id=instance_id,
old_flavor=self._transform_obj(old_flavor),
new_flavor=self._transform_obj(new_flavor))
def reboot(self, instance_id):
LOG.debug("Making async call to reboot instance: %s" % instance_id)
version = self.API_BASE_VERSION
self._cast("reboot", self.version_cap, instance_id=instance_id)
self._cast("reboot", version=version, instance_id=instance_id)
def restart(self, instance_id):
LOG.debug("Making async call to restart instance: %s" % instance_id)
version = self.API_BASE_VERSION
self._cast("restart", self.version_cap, instance_id=instance_id)
self._cast("restart", version=version, instance_id=instance_id)
def detach_replica(self, instance_id):
LOG.debug("Making async call to detach replica: %s" % instance_id)
version = self.API_BASE_VERSION
self._cast("detach_replica", self.version_cap,
self._cast("detach_replica", version=version,
instance_id=instance_id)
def promote_to_replica_source(self, instance_id):
LOG.debug("Making async call to promote replica to source: %s" %
instance_id)
self._cast("promote_to_replica_source", self.version_cap,
version = self.API_BASE_VERSION
self._cast("promote_to_replica_source", version=version,
instance_id=instance_id)
def eject_replica_source(self, instance_id):
LOG.debug("Making async call to eject replica source: %s" %
instance_id)
self._cast("eject_replica_source", self.version_cap,
version = self.API_BASE_VERSION
self._cast("eject_replica_source", version=version,
instance_id=instance_id)
def migrate(self, instance_id, host):
LOG.debug("Making async call to migrate instance: %s" % instance_id)
version = self.API_BASE_VERSION
self._cast("migrate", self.version_cap,
self._cast("migrate", version=version,
instance_id=instance_id, host=host)
def delete_instance(self, instance_id):
LOG.debug("Making async call to delete instance: %s" % instance_id)
version = self.API_BASE_VERSION
self._cast("delete_instance", self.version_cap,
self._cast("delete_instance", version=version,
instance_id=instance_id)
def create_backup(self, backup_info, instance_id):
LOG.debug("Making async call to create a backup for instance: %s" %
instance_id)
version = self.API_BASE_VERSION
self._cast("create_backup", self.version_cap,
self._cast("create_backup", version=version,
backup_info=backup_info,
instance_id=instance_id)
def delete_backup(self, backup_id):
LOG.debug("Making async call to delete backup: %s" % backup_id)
version = self.API_BASE_VERSION
self._cast("delete_backup", self.version_cap, backup_id=backup_id)
self._cast("delete_backup", version=version, backup_id=backup_id)
def create_instance(self, instance_id, name, flavor,
image_id, databases, users, datastore_manager,
@ -155,7 +189,8 @@ class API(object):
modules=None, locality=None):
LOG.debug("Making async call to create instance %s " % instance_id)
self._cast("create_instance", self.version_cap,
version = self.API_BASE_VERSION
self._cast("create_instance", version=version,
instance_id=instance_id, name=name,
flavor=self._transform_obj(flavor),
image_id=image_id,
@ -176,33 +211,38 @@ class API(object):
def create_cluster(self, cluster_id):
LOG.debug("Making async call to create cluster %s " % cluster_id)
version = self.API_BASE_VERSION
self._cast("create_cluster", self.version_cap, cluster_id=cluster_id)
self._cast("create_cluster", version=version, cluster_id=cluster_id)
def grow_cluster(self, cluster_id, new_instance_ids):
LOG.debug("Making async call to grow cluster %s " % cluster_id)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=self.version_cap)
cctxt = self.client.prepare(version=version)
cctxt.cast(self.context, "grow_cluster",
cluster_id=cluster_id, new_instance_ids=new_instance_ids)
def shrink_cluster(self, cluster_id, instance_ids):
LOG.debug("Making async call to shrink cluster %s " % cluster_id)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=self.version_cap)
cctxt = self.client.prepare(version=version)
cctxt.cast(self.context, "shrink_cluster",
cluster_id=cluster_id, instance_ids=instance_ids)
def delete_cluster(self, cluster_id):
LOG.debug("Making async call to delete cluster %s " % cluster_id)
version = self.API_BASE_VERSION
self._cast("delete_cluster", self.version_cap, cluster_id=cluster_id)
self._cast("delete_cluster", version=version, cluster_id=cluster_id)
def upgrade(self, instance_id, datastore_version_id):
LOG.debug("Making async call to upgrade guest to datastore "
"version %s " % datastore_version_id)
version = self.API_BASE_VERSION
cctxt = self.client.prepare(version=self.version_cap)
cctxt = self.client.prepare(version=version)
cctxt.cast(self.context, "upgrade", instance_id=instance_id,
datastore_version_id=datastore_version_id)

4
trove/taskmanager/manager.py
View File

@ -14,7 +14,6 @@
# under the License.
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_service import periodic_task
from oslo_utils import importutils
@ -27,7 +26,6 @@ from trove.common.exception import TroveError
from trove.common.i18n import _
from trove.common.notification import DBaaSQuotas, EndNotification
from trove.common import remote
import trove.common.rpc.version as rpc_version
from trove.common import server_group as srv_grp
from trove.common.strategies.cluster import strategy
from trove.datastore.models import DatastoreVersion
@ -43,8 +41,6 @@ CONF = cfg.CONF
class Manager(periodic_task.PeriodicTasks):
target = messaging.Target(version=rpc_version.RPC_API_VERSION)
def __init__(self):
super(Manager, self).__init__(CONF)
self.admin_context = TroveContext(