openstack
/
magnum
Code Issues Proposed changes

[WIP] ng-8: APIs for nodegroup CRUD operations 

This adds the changes needed in the API and conductor level to support
creating updating and deleting nodegroups.

Change-Id: I4ad60994ad6b4cb9cac18129557e1e87e61ae98c
This commit is contained in:
Theodoros Tsioutsias 2019-03-26 11:28:59 +00:00
parent 1f7d4206cf
commit ee8fa94471
9 changed files with 390 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
magnum/api/controllers/v1/cluster.py
View File

@ -397,22 +397,26 @@ class ClustersController(base.Controller):
and store them into cluster.faults.
"""
osc = clients.OpenStackClients(context)
filters = {'status': 'FAILED'}
try:
failed_resources = osc.heat().resources.list(
cluster.stack_id, nested_depth=2, filters=filters)
except Exception as e:
failed_resources = []
LOG.warning("Failed to retrieve failed resources for "
"cluster %(cluster)s from Heat stack "
"%(stack)s due to error: %(e)s",
{'cluster': cluster.uuid,
'stack': cluster.stack_id, 'e': e},
exc_info=True)
return {
ng.name: ng.status_reason for ng in cluster.nodegroups
if ng.status.endswith('FAILED')
}
# osc = clients.OpenStackClients(context)
# filters = {'status': 'FAILED'}
# try:
# failed_resources = osc.heat().resources.list(
# cluster.stack_id, nested_depth=2, filters=filters)
# except Exception as e:
# failed_resources = []
# LOG.warning("Failed to retrieve failed resources for "
# "cluster %(cluster)s from Heat stack "
# "%(stack)s due to error: %(e)s",
# {'cluster': cluster.uuid,
# 'stack': cluster.stack_id, 'e': e},
# exc_info=True)
return {res.resource_name: res.resource_status_reason
for res in failed_resources}
# return {res.resource_name: res.resource_status_reason
# for res in failed_resources}
@expose.expose(Cluster, types.uuid_or_name)
def get_one(self, cluster_ident):

135
magnum/api/controllers/v1/nodegroup.py
View File

@ -14,6 +14,7 @@
# under the License.
import pecan
import uuid
import wsme
from wsme import types as wtypes
@ -23,8 +24,10 @@ from magnum.api.controllers.v1 import collection
from magnum.api.controllers.v1 import types
from magnum.api import expose
from magnum.api import utils as api_utils
from magnum.common import exception
from magnum.common import policy
from magnum import objects
from magnum.objects import fields
class NodeGroup(base.APIBase):
@ -82,6 +85,18 @@ class NodeGroup(base.APIBase):
is_default = types.BooleanType()
"""Specifies is a nodegroup was created by default or not"""
stack_id = wsme.wsattr(wtypes.text, readonly=True)
"""Stack id of the heat stack"""
status = wtypes.Enum(wtypes.text, *fields.ClusterStatus.ALL)
"""Status of the nodegroup from the heat stack"""
status_reason = wtypes.text
"""Status reason of the nodegroup from the heat stack"""
version = wtypes.text
"""Version of the nodegroup"""
def __init__(self, **kwargs):
super(NodeGroup, self).__init__()
self.fields = []
@ -101,7 +116,8 @@ class NodeGroup(base.APIBase):
ng = NodeGroup(**nodegroup.as_dict())
if not expand:
ng.unset_fields_except(["uuid", "name", "flavor_id", "node_count",
"role", "is_default", "image_id"])
"role", "is_default", "image_id", "status",
"stack_id"])
else:
ng.links = [link.Link.make_link('self', url, cluster_path,
nodegroup_path),
@ -111,6 +127,20 @@ class NodeGroup(base.APIBase):
return ng
class NodeGroupPatchType(types.JsonPatchType):
_api_base = NodeGroup
@staticmethod
def internal_attrs():
# Allow updating only min/max_node_count
internal_attrs = ["/name", "/cluster_id", "/project_id",
"/docker_volume_size", "/labels", "/flavor_id",
"/image_id", "/node_addresses", "/node_count",
"/role", "/is_default", "/stack_id", "/status",
"/status_reason", "/version"]
return types.JsonPatchType.internal_attrs() + internal_attrs
class NodeGroupCollection(collection.Collection):
"""API representation of a collection of Node Groups."""
@ -217,3 +247,106 @@ class NodeGroupController(base.Controller):
cluster = api_utils.get_resource('Cluster', cluster_id)
nodegroup = objects.NodeGroup.get(context, cluster.uuid, nodegroup_id)
return NodeGroup.convert(nodegroup)
@expose.expose(NodeGroup, types.uuid_or_name, NodeGroup, body=NodeGroup,
status_code=201)
def post(self, cluster_id, nodegroup):
"""Create NodeGroup.
:param nodegroup: a json document to create this NodeGroup.
"""
context = pecan.request.context
policy.enforce(context, 'nodegroup:create', action='nodegroup:create')
cluster = api_utils.get_resource('Cluster', cluster_id)
cluster_ngs = [ng.name for ng in cluster.nodegroups]
if nodegroup.name in cluster_ngs:
raise exception.NodeGroupAlreadyExists(name=nodegroup.name,
cluster_id=cluster.name)
if (nodegroup.node_count < nodegroup.min_node_count or
nodegroup.node_count > nodegroup.max_node_count):
allowed = ("min_node_count=%s, max_node_count=%s"
% (nodegroup.min_node_count, nodegroup.max_node_count))
raise exception.NodeGroupInvalidInput(attr='node_count',
nodegroup=nodegroup.name,
allowed=allowed)
if not nodegroup.image_id:
nodegroup.image_id = cluster.cluster_template.image_id
if not nodegroup.flavor_id:
nodegroup.flavor_id = cluster.flavor_id
if nodegroup.labels == wtypes.Unset:
nodegroup.labels = cluster.labels
# set this to minion explicitly
nodegroup.role = "worker"
if nodegroup.labels == wtypes.Unset:
nodegroup.labels = cluster.labels
nodegroup_dict = nodegroup.as_dict()
nodegroup_dict['cluster_id'] = cluster.uuid
nodegroup_dict['project_id'] = context.project_id
nodegroup_dict['user_id'] = context.user_id
nodegroup_dict['coe_version'] = cluster.coe_version
nodegroup_dict['container_version'] = cluster.container_version
new_obj = objects.NodeGroup(context, **nodegroup_dict)
new_obj.uuid = uuid.uuid4()
pecan.request.rpcapi.nodegroup_create_async(cluster, new_obj)
return NodeGroup.convert(new_obj)
@expose.expose(NodeGroup, types.uuid_or_name, types.uuid_or_name,
body=[NodeGroupPatchType], status_code=202)
def patch(self, cluster_id, nodegroup_id, patch):
"""Update NodeGroup.
:param cluster_id: cluster id.
:param : resource name.
:param values: a json document to update a nodegroup.
"""
cluster = api_utils.get_resource('Cluster', cluster_id)
nodegroup = self._patch(cluster.uuid, nodegroup_id, patch)
pecan.request.rpcapi.nodegroup_update_async(cluster, nodegroup)
return NodeGroup.convert(nodegroup)
@expose.expose(None, types.uuid_or_name, types.uuid_or_name,
status_code=204)
def delete(self, cluster_id, nodegroup_id):
"""Delete NodeGroup for a given project_id and resource.
:param cluster_id: cluster id.
:param nodegroup_id: resource name.
"""
context = pecan.request.context
policy.enforce(context, 'nodegroup:delete', action='nodegroup:delete')
cluster = api_utils.get_resource('Cluster', cluster_id)
nodegroup = objects.NodeGroup.get(context, cluster.uuid, nodegroup_id)
if nodegroup.is_default:
raise exception.DeletingDefaultNGNotSupported()
pecan.request.rpcapi.nodegroup_delete_async(cluster, nodegroup)
def _patch(self, cluster_uuid, nodegroup_id, patch):
context = pecan.request.context
policy.enforce(context, 'nodegroup:update', action='nodegroup:update')
nodegroup = objects.NodeGroup.get(context, cluster_uuid, nodegroup_id)
try:
ng_dict = nodegroup.as_dict()
new_nodegroup = NodeGroup(**api_utils.apply_jsonpatch(ng_dict,
patch))
except api_utils.JSONPATCH_EXCEPTIONS as e:
raise exception.PatchError(patch=patch, reason=e)
# Update only the fields that have changed
for field in objects.NodeGroup.fields:
try:
patch_val = getattr(new_nodegroup, field)
except AttributeError:
# Ignore fields that aren't exposed in the API
continue
if patch_val == wtypes.Unset:
patch_val = None
if nodegroup[field] != patch_val:
nodegroup[field] = patch_val
return nodegroup

2
magnum/cmd/conductor.py
View File

@ -30,6 +30,7 @@ from magnum.conductor.handlers import cluster_conductor
from magnum.conductor.handlers import conductor_listener
from magnum.conductor.handlers import federation_conductor
from magnum.conductor.handlers import indirection_api
from magnum.conductor.handlers import nodegroup_conductor
import magnum.conf
from magnum import version
@ -53,6 +54,7 @@ def main():
conductor_listener.Handler(),
ca_conductor.Handler(),
federation_conductor.Handler(),
nodegroup_conductor.Handler(),
]
server = rpc_service.Service.create(CONF.conductor.topic,

8
magnum/common/exception.py
View File

@ -433,3 +433,11 @@ class NGResizeOutBounds(Invalid):
message = _("Resizing %(nodegroup)s outside the allowed range: "
"min_node_count = %(min_node_count)s, "
"max_node_count = %(max_node_count)s")
class DeletingDefaultNGNotSupported(NotSupported):
message = _("Deleting a default nodegroup is not supported.")
class NodeGroupInvalidInput(Conflict):
message = _("%(attr)s for %(nodegroup)s is invalid (%(allowed)s).")

33
magnum/common/policies/nodegroup.py
View File

@ -66,6 +66,39 @@ rules = [
}
]
),
policy.DocumentedRuleDefault(
name=NODEGROUP % 'create',
check_str=base.RULE_ADMIN_OR_OWNER,
description='Create a new nodegroup.',
operations=[
{
'path': '/v1/clusters/{cluster_id}/nodegroups/',
'method': 'POST'
}
]
),
policy.DocumentedRuleDefault(
name=NODEGROUP % 'delete',
check_str=base.RULE_ADMIN_OR_OWNER,
description='Delete a nodegroup.',
operations=[
{
'path': '/v1/clusters/{cluster_id}/nodegroups/{nodegroup}',
'method': 'DELETE'
}
]
),
policy.DocumentedRuleDefault(
name=NODEGROUP % 'update',
check_str=base.RULE_ADMIN_OR_OWNER,
description='Update an existing nodegroup.',
operations=[
{
'path': '/v1/clusters/{cluster_id}/nodegroups/{nodegroup}',
'method': 'PATCH'
}
]
),
]

23
magnum/conductor/api.py
View File

@ -144,6 +144,29 @@ class API(rpc_service.API):
return self._client.call(context, 'object_backport', objinst=objinst,
target_version=target_version)
# NodeGroup Operations
def nodegroup_create(self, cluster, nodegroup):
return self._call('nodegroup_create', cluster=cluster,
nodegroup=nodegroup)
def nodegroup_create_async(self, cluster, nodegroup):
self._cast('nodegroup_create', cluster=cluster, nodegroup=nodegroup)
def nodegroup_delete(self, cluster, nodegroup):
return self._call('nodegroup_delete', cluster=cluster,
nodegroup=nodegroup)
def nodegroup_delete_async(self, cluster, nodegroup):
self._cast('nodegroup_delete', cluster=cluster, nodegroup=nodegroup)
def nodegroup_update(self, cluster, nodegroup):
return self._call('nodegroup_update', cluster=cluster,
nodegroup=nodegroup)
def nodegroup_update_async(self, cluster, nodegroup):
self._cast('nodegroup_update', cluster=cluster, nodegroup=nodegroup)
@profiler.trace_cls("rpc")
class ListenerAPI(rpc_service.API):

147
magnum/conductor/handlers/nodegroup_conductor.py Normal file
View File

@ -0,0 +1,147 @@
# Copyright (c) 2018 European Organization for Nuclear Research.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import functools
from heatclient import exc
from oslo_log import log as logging
import six
from magnum.common import exception
from magnum.common import profiler
import magnum.conf
from magnum.drivers.common import driver
from magnum.i18n import _
from magnum.objects import fields
CONF = magnum.conf.CONF
LOG = logging.getLogger(__name__)
# TODO(ttsiouts): notifications about nodegroup operations will be
# added in later commit.
ALLOWED_CLUSTER_STATES = (
fields.ClusterStatus.CREATE_COMPLETE,
fields.ClusterStatus.UPDATE_COMPLETE,
fields.ClusterStatus.UPDATE_IN_PROGRESS,
fields.ClusterStatus.UPDATE_FAILED,
fields.ClusterStatus.RESUME_COMPLETE,
fields.ClusterStatus.RESTORE_COMPLETE,
fields.ClusterStatus.ROLLBACK_COMPLETE,
fields.ClusterStatus.SNAPSHOT_COMPLETE,
fields.ClusterStatus.CHECK_COMPLETE,
fields.ClusterStatus.ADOPT_COMPLETE
)
def allowed_operation(func):
@functools.wraps(func)
def wrapper(self, context, cluster, nodegroup, *args, **kwargs):
# Before we begin we need to check the status
# of the cluster. If the cluster is in a status
# that does not allow nodegroup creation we just
# fail.
if cluster.status not in ALLOWED_CLUSTER_STATES:
operation = _(
'%(fname)s when cluster status is "%(status)s"'
) % {'fname': func.__name__, 'status': cluster.status}
raise exception.NotSupported(operation=operation)
return func(self, context, cluster, nodegroup, *args, **kwargs)
return wrapper
@profiler.trace_cls("rpc")
class Handler(object):
@allowed_operation
def nodegroup_create(self, context, cluster, nodegroup):
LOG.debug("nodegroup_conductor nodegroup_create")
cluster.status = fields.ClusterStatus.UPDATE_IN_PROGRESS
cluster.save()
nodegroup.create()
try:
cluster_driver = driver.Driver.get_driver_for_cluster(context,
cluster)
cluster_driver.create_nodegroup(context, cluster, nodegroup)
nodegroup.save()
except Exception as e:
nodegroup.status = fields.ClusterStatus.CREATE_FAILED
nodegroup.status_reason = six.text_type(e)
nodegroup.save()
cluster.status = fields.ClusterStatus.UPDATE_FAILED
cluster.status_reason = six.text_type(e)
cluster.save()
nodegroup.save()
if isinstance(e, exc.HTTPBadRequest):
e = exception.InvalidParameterValue(message=six.text_type(e))
raise e
raise
return nodegroup
@allowed_operation
def nodegroup_update(self, context, cluster, nodegroup):
LOG.debug("nodegroup_conductor nodegroup_update")
cluster.status = fields.ClusterStatus.UPDATE_IN_PROGRESS
cluster.save()
try:
cluster_driver = driver.Driver.get_driver_for_cluster(context,
cluster)
cluster_driver.update_nodegroup(context, cluster, nodegroup)
nodegroup.save()
except Exception as e:
nodegroup.status = fields.ClusterStatus.UPDATE_FAILED
nodegroup.status_reason = six.text_type(e)
nodegroup.save()
cluster.status = fields.ClusterStatus.UPDATE_FAILED
cluster.status_reason = six.text_type(e)
cluster.save()
nodegroup.save()
if isinstance(e, exc.HTTPBadRequest):
e = exception.InvalidParameterValue(message=six.text_type(e))
raise e
raise
return nodegroup
def nodegroup_delete(self, context, cluster, nodegroup):
LOG.debug("nodegroup_conductor nodegroup_delete")
cluster.status = fields.ClusterStatus.UPDATE_IN_PROGRESS
cluster.save()
try:
cluster_driver = driver.Driver.get_driver_for_cluster(context,
cluster)
cluster_driver.delete_nodegroup(context, cluster, nodegroup)
nodegroup.save()
except Exception as e:
nodegroup.status = fields.ClusterStatus.DELETE_FAILED
nodegroup.status_reason = six.text_type(e)
nodegroup.save()
cluster.status = fields.ClusterStatus.UPDATE_FAILED
cluster.status_reason = six.text_type(e)
cluster.save()
nodegroup.save()
if isinstance(e, exc.HTTPBadRequest):
e = exception.InvalidParameterValue(message=six.text_type(e))
raise e
raise
return None

15
magnum/drivers/common/driver.py
View File

@ -209,6 +209,21 @@ class Driver(object):
raise NotImplementedError("Subclasses must implement "
"'delete_federation'.")
@abc.abstractmethod
def create_nodegroup(self, context, cluster, nodegroup):
raise NotImplementedError("Subclasses must implement "
"'create_nodegroup'.")
@abc.abstractmethod
def update_nodegroup(self, context, cluster, nodegroup):
raise NotImplementedError("Subclasses must implement "
"'update_nodegroup'.")
@abc.abstractmethod
def delete_nodegroup(self, context, cluster, nodegroup):
raise NotImplementedError("Subclasses must implement "
"'delete_nodegroup'.")
def get_monitor(self, context, cluster):
"""return the monitor with container data for this driver."""

9
magnum/drivers/heat/driver.py
View File

@ -96,6 +96,15 @@ class HeatDriver(driver.Driver):
def delete_federation(self, context, federation):
return NotImplementedError("Must implement 'delete_federation'")
def create_nodegroup(self, context, cluster, nodegroup):
raise NotImplementedError("Must implement 'create_nodegroup'.")
def update_nodegroup(self, context, cluster, nodegroup):
raise NotImplementedError("Must implement 'update_nodegroup'.")
def delete_nodegroup(self, context, cluster, nodegroup):
raise NotImplementedError("Must implement 'delete_nodegroup'.")
def update_cluster_status(self, context, cluster):
if cluster.stack_id is None:
# NOTE(mgoddard): During cluster creation it is possible to poll