Refactor MEO's db

Change-Id: I27c7f2513b45e93bdc859a900bc540589e3a43d3
Signed-off-by: doantungbk <doantungbk.203@gmail.com>
This commit is contained in:
doantungbk 2018-10-29 11:33:05 -07:00
parent 81fa39222d
commit 8f4d8fe306
12 changed files with 893 additions and 630 deletions

451
apmec/db/meo/meca_db.py Normal file
View File

@ -0,0 +1,451 @@
# 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 ast
from datetime import datetime
from oslo_db.exception import DBDuplicateEntry
from oslo_log import log as logging
from oslo_utils import timeutils
from oslo_utils import uuidutils
from six import iteritems
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy import schema
from apmec.common import exceptions
from apmec.db.common_services import common_services_db_plugin
from apmec.db import db_base
from apmec.db import model_base
from apmec.db import models_v1
from apmec.db import types
from apmec.extensions import meo
from apmec.plugins.common import constants
LOG = logging.getLogger(__name__)
_ACTIVE_UPDATE = (constants.ACTIVE, constants.PENDING_UPDATE)
_ACTIVE_UPDATE_ERROR_DEAD = (
constants.PENDING_CREATE, constants.ACTIVE, constants.PENDING_UPDATE,
constants.ERROR, constants.DEAD)
CREATE_STATES = (constants.PENDING_CREATE, constants.DEAD)
###########################################################################
# db tables
class MECAD(model_base.BASE, models_v1.HasId, models_v1.HasTenant,
models_v1.Audit):
"""Represents MECAD to create MECA."""
__tablename__ = 'mecad'
# Descriptive name
name = sa.Column(sa.String(255), nullable=False)
description = sa.Column(sa.Text)
meads = sa.Column(types.Json, nullable=True)
# Mecad template source - onboarded
template_source = sa.Column(sa.String(255), server_default='onboarded')
# (key, value) pair to spin up
attributes = orm.relationship('MECADAttribute',
backref='mecad')
__table_args__ = (
schema.UniqueConstraint(
"tenant_id",
"name",
name="uniq_mecad0tenant_id0name"),
)
class MECADAttribute(model_base.BASE, models_v1.HasId):
"""Represents attributes necessary for creation of meca in (key, value) pair
"""
__tablename__ = 'mecad_attribute'
mecad_id = sa.Column(types.Uuid, sa.ForeignKey('mecad.id'),
nullable=False)
key = sa.Column(sa.String(255), nullable=False)
value = sa.Column(sa.TEXT(65535), nullable=True)
class MECA(model_base.BASE, models_v1.HasId, models_v1.HasTenant,
models_v1.Audit):
"""Represents network services that deploys services.
"""
__tablename__ = 'meca'
mecad_id = sa.Column(types.Uuid, sa.ForeignKey('mecad.id'))
mecad = orm.relationship('MECAD')
name = sa.Column(sa.String(255), nullable=False)
description = sa.Column(sa.Text, nullable=True)
# Dict of MEA details that network service launches
mea_ids = sa.Column(sa.TEXT(65535), nullable=True)
# Dict of mgmt urls that network servic launches
mgmt_urls = sa.Column(sa.TEXT(65535), nullable=True)
status = sa.Column(sa.String(64), nullable=False)
vim_id = sa.Column(types.Uuid, sa.ForeignKey('vims.id'), nullable=False)
error_reason = sa.Column(sa.Text, nullable=True)
__table_args__ = (
schema.UniqueConstraint(
"tenant_id",
"name",
name="uniq_meca0tenant_id0name"),
)
class MECAPluginDb(meo.MECAPluginBase, db_base.CommonDbMixin):
def __init__(self):
super(MECAPluginDb, self).__init__()
self._cos_db_plg = common_services_db_plugin.CommonServicesPluginDb()
def _get_resource(self, context, model, id):
try:
return self._get_by_id(context, model, id)
except orm_exc.NoResultFound:
if issubclass(model, MECAD):
raise meo.MECADNotFound(mecad_id=id)
if issubclass(model, MECA):
raise meo.MECANotFound(meca_id=id)
else:
raise
def _get_meca_db(self, context, meca_id, current_statuses, new_status):
try:
meca_db = (
self._model_query(context, MECA).
filter(MECA.id == meca_id).
filter(MECA.status.in_(current_statuses)).
with_lockmode('update').one())
except orm_exc.NoResultFound:
raise meo.MECANotFound(meca_id=meca_id)
meca_db.update({'status': new_status})
return meca_db
def _make_attributes_dict(self, attributes_db):
return dict((attr.key, attr.value) for attr in attributes_db)
def _make_mecad_dict(self, mecad, fields=None):
res = {
'attributes': self._make_attributes_dict(mecad['attributes']),
}
key_list = ('id', 'tenant_id', 'name', 'description',
'created_at', 'updated_at', 'meads', 'template_source')
res.update((key, mecad[key]) for key in key_list)
return self._fields(res, fields)
def _make_dev_attrs_dict(self, dev_attrs_db):
return dict((arg.key, arg.value) for arg in dev_attrs_db)
def _make_meca_dict(self, meca_db, fields=None):
LOG.debug('meca_db %s', meca_db)
res = {}
key_list = ('id', 'tenant_id', 'mecad_id', 'name', 'description',
'mea_ids', 'status', 'mgmt_urls', 'error_reason',
'vim_id', 'created_at', 'updated_at')
res.update((key, meca_db[key]) for key in key_list)
return self._fields(res, fields)
def create_mecad(self, context, mecad):
meads = mecad['meads']
mecad = mecad['mecad']
LOG.debug('mecad %s', mecad)
tenant_id = self._get_tenant_id_for_create(context, mecad)
template_source = mecad.get('template_source')
try:
with context.session.begin(subtransactions=True):
mecad_id = uuidutils.generate_uuid()
mecad_db = MECAD(
id=mecad_id,
tenant_id=tenant_id,
name=mecad.get('name'),
meads=meads,
description=mecad.get('description'),
deleted_at=datetime.min,
template_source=template_source)
context.session.add(mecad_db)
for (key, value) in mecad.get('attributes', {}).items():
attribute_db = MECADAttribute(
id=uuidutils.generate_uuid(),
mecad_id=mecad_id,
key=key,
value=value)
context.session.add(attribute_db)
except DBDuplicateEntry as e:
raise exceptions.DuplicateEntity(
_type="mecad",
entry=e.columns)
LOG.debug('mecad_db %(mecad_db)s %(attributes)s ',
{'mecad_db': mecad_db,
'attributes': mecad_db.attributes})
mecad_dict = self._make_mecad_dict(mecad_db)
LOG.debug('mecad_dict %s', mecad_dict)
self._cos_db_plg.create_event(
context, res_id=mecad_dict['id'],
res_type=constants.RES_TYPE_MECAD,
res_state=constants.RES_EVT_ONBOARDED,
evt_type=constants.RES_EVT_CREATE,
tstamp=mecad_dict[constants.RES_EVT_CREATED_FLD])
return mecad_dict
def delete_mecad(self,
context,
mecad_id,
soft_delete=True):
with context.session.begin(subtransactions=True):
mecas_db = context.session.query(MECA).filter_by(
mecad_id=mecad_id).first()
if mecas_db is not None and mecas_db.deleted_at is None:
raise meo.MECADInUse(mecad_id=mecad_id)
mecad_db = self._get_resource(context, MECAD,
mecad_id)
if soft_delete:
mecad_db.update({'deleted_at': timeutils.utcnow()})
self._cos_db_plg.create_event(
context, res_id=mecad_db['id'],
res_type=constants.RES_TYPE_MECAD,
res_state=constants.RES_EVT_NA_STATE,
evt_type=constants.RES_EVT_DELETE,
tstamp=mecad_db[constants.RES_EVT_DELETED_FLD])
else:
context.session.query(MECADAttribute).filter_by(
mecad_id=mecad_id).delete()
context.session.delete(mecad_db)
def get_mecad(self, context, mecad_id, fields=None):
mecad_db = self._get_resource(context, MECAD, mecad_id)
return self._make_mecad_dict(mecad_db)
def get_mecads(self, context, filters, fields=None):
if ('template_source' in filters) and \
(filters['template_source'][0] == 'all'):
filters.pop('template_source')
return self._get_collection(context, MECAD,
self._make_mecad_dict,
filters=filters, fields=fields)
# reference implementation. needs to be overrided by subclass
def create_meca(self, context, meca):
LOG.debug('meca %s', meca)
meca = meca['meca']
tenant_id = self._get_tenant_id_for_create(context, meca)
mecad_id = meca['mecad_id']
vim_id = meca['vim_id']
name = meca.get('name')
meca_id = uuidutils.generate_uuid()
try:
with context.session.begin(subtransactions=True):
mecad_db = self._get_resource(context, MECAD,
mecad_id)
meca_db = MECA(id=meca_id,
tenant_id=tenant_id,
name=name,
description=mecad_db.description,
mea_ids=None,
status=constants.PENDING_CREATE,
mgmt_urls=None,
mecad_id=mecad_id,
vim_id=vim_id,
error_reason=None,
deleted_at=datetime.min)
context.session.add(meca_db)
except DBDuplicateEntry as e:
raise exceptions.DuplicateEntity(
_type="meca",
entry=e.columns)
evt_details = "MECA UUID assigned."
self._cos_db_plg.create_event(
context, res_id=meca_id,
res_type=constants.RES_TYPE_meca,
res_state=constants.PENDING_CREATE,
evt_type=constants.RES_EVT_CREATE,
tstamp=meca_db[constants.RES_EVT_CREATED_FLD],
details=evt_details)
return self._make_meca_dict(meca_db)
def create_meca_post(self, context, meca_id, mistral_obj,
mead_dict, error_reason):
LOG.debug('meca ID %s', meca_id)
output = ast.literal_eval(mistral_obj.output)
mgmt_urls = dict()
mea_ids = dict()
if len(output) > 0:
for mead_name, mead_val in iteritems(mead_dict):
for instance in mead_val['instances']:
if 'mgmt_url_' + instance in output:
mgmt_url_dict =\
ast.literal_eval(
output['mgmt_url_' + instance].strip())
mgmt_urls[instance] = mgmt_url_dict.values()
mea_ids[instance] = list()
mea_ids[instance].append(output['mea_id_' + instance])
mea_ids = str(mea_ids)
mgmt_urls = str(mgmt_urls)
if not mea_ids:
mea_ids = None
if not mgmt_urls:
mgmt_urls = None
status = constants.ACTIVE if mistral_obj.state == 'SUCCESS' \
else constants.ERROR
with context.session.begin(subtransactions=True):
meca_db = self._get_resource(context, MECA,
meca_id)
meca_db.update({'mea_ids': mea_ids})
meca_db.update({'mgmt_urls': mgmt_urls})
meca_db.update({'status': status})
meca_db.update({'error_reason': error_reason})
meca_db.update({'updated_at': timeutils.utcnow()})
meca_dict = self._make_meca_dict(meca_db)
self._cos_db_plg.create_event(
context, res_id=meca_dict['id'],
res_type=constants.RES_TYPE_meca,
res_state=constants.RES_EVT_NA_STATE,
evt_type=constants.RES_EVT_UPDATE,
tstamp=meca_dict[constants.RES_EVT_UPDATED_FLD])
return meca_dict
# reference implementation. needs to be overrided by subclass
def delete_meca(self, context, meca_id):
with context.session.begin(subtransactions=True):
meca_db = self._get_meca_db(
context, meca_id, _ACTIVE_UPDATE_ERROR_DEAD,
constants.PENDING_DELETE)
deleted_meca_db = self._make_meca_dict(meca_db)
self._cos_db_plg.create_event(
context, res_id=meca_id,
res_type=constants.RES_TYPE_meca,
res_state=deleted_meca_db['status'],
evt_type=constants.RES_EVT_DELETE,
tstamp=timeutils.utcnow(), details="MECA delete initiated")
return deleted_meca_db
def delete_meca_post(self, context, meca_id, mistral_obj,
error_reason, soft_delete=True):
meca = self.get_meca(context, meca_id)
mecad_id = meca.get('mecad_id')
with context.session.begin(subtransactions=True):
query = (
self._model_query(context, MECA).
filter(MECA.id == meca_id).
filter(MECA.status == constants.PENDING_DELETE))
if mistral_obj and mistral_obj.state == 'ERROR':
query.update({'status': constants.ERROR})
self._cos_db_plg.create_event(
context, res_id=meca_id,
res_type=constants.RES_TYPE_meca,
res_state=constants.ERROR,
evt_type=constants.RES_EVT_DELETE,
tstamp=timeutils.utcnow(),
details="MECA Delete ERROR")
else:
if soft_delete:
deleted_time_stamp = timeutils.utcnow()
query.update({'deleted_at': deleted_time_stamp})
self._cos_db_plg.create_event(
context, res_id=meca_id,
res_type=constants.RES_TYPE_meca,
res_state=constants.PENDING_DELETE,
evt_type=constants.RES_EVT_DELETE,
tstamp=deleted_time_stamp,
details="meca Delete Complete")
else:
query.delete()
template_db = self._get_resource(context, MECAD, mecad_id)
if template_db.get('template_source') == 'inline':
self.delete_mecad(context, mecad_id)
def get_meca(self, context, meca_id, fields=None):
meca_db = self._get_resource(context, MECA, meca_id)
return self._make_meca_dict(meca_db)
def get_mecas(self, context, filters=None, fields=None):
return self._get_collection(context, MECA,
self._make_meca_dict,
filters=filters, fields=fields)
def _update_meca_pre(self, context, meca_id):
with context.session.begin(subtransactions=True):
meca_db = self._get_meca_db(
context, meca_id, _ACTIVE_UPDATE,
constants.PENDING_UPDATE)
return self._make_meca_dict(meca_db)
def _update_meca_post(self, context, meca_id, mistral_obj,
mead_dict, error_reason):
output = ast.literal_eval(mistral_obj.output)
new_mgmt_urls = dict()
new_mea_ids = dict()
if len(output) > 0:
for mead_name, mead_val in iteritems(mead_dict):
for instance in mead_val['instances']:
if 'mgmt_url_' + instance in output:
mgmt_url_dict = ast.literal_eval(
output['mgmt_url_' + instance].strip())
new_mgmt_urls[instance] = mgmt_url_dict.values()
new_mea_ids[instance] = output['mea_id_' + instance]
if not new_mea_ids:
new_mea_ids = None
if not new_mgmt_urls:
new_mgmt_urls = None
status = constants.ACTIVE if mistral_obj.state == 'SUCCESS' \
else constants.ERROR
with context.session.begin(subtransactions=True):
meca_db = self._get_resource(context, MECA, meca_id)
mgmt_urls = ast.literal_eval(meca_db.mgmt_urls)
for mea_name, mgmt_dict in mgmt_urls.items():
for new_mea_name, new_mgmt_dict in new_mgmt_urls.items():
if new_mea_name == mea_name:
extra_mgmt = new_mgmt_urls.pop(new_mea_name)
mgmt_urls[mea_name].extend(extra_mgmt)
mgmt_urls.update(new_mgmt_urls)
mgmt_urls = str(mgmt_urls)
mea_ids = ast.literal_eval(meca_db.mea_ids)
for mea_name, mea_id_list in mea_ids.items():
for new_mea_name, new_mead_id_list in new_mea_ids.items():
if new_mea_name == mea_name:
extra_id = new_mea_ids.pop(new_mea_name)
mea_ids[mea_name].append(extra_id)
mea_ids.update(new_mea_ids)
mea_ids = str(mea_ids)
meca_db.update({'mea_ids': mea_ids})
meca_db.update({'mgmt_urls': mgmt_urls})
meca_db.update({'status': status})
meca_db.update({'error_reason': error_reason})
meca_db.update({'updated_at': timeutils.utcnow()})
meca_dict = self._make_meca_dict(meca_db)
return meca_dict
def _update_meca_status(self, context, meca_id, new_status):
with context.session.begin(subtransactions=True):
meca_db = self._get_meca_db(
context, meca_id, _ACTIVE_UPDATE, new_status)
return self._make_meca_dict(meca_db)
def update_meca(self, context, meca_id, meca):
meca_dict = self._update_meca_pre(context, meca_id)
self._update_meca_post(context, meca_id,
constants.ACTIVE, meca_dict, None)

View File

@ -10,15 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
from apmec.common import exceptions
from apmec.db.common_services import common_services_db_plugin
from apmec.db import db_base
from apmec.db.mem import mem_db
from apmec.db.meo import meo_db
from apmec.extensions import meo
from apmec import manager
from apmec.plugins.common import constants
from datetime import datetime
from oslo_db.exception import DBDuplicateEntry
@ -28,6 +19,15 @@ from oslo_utils import uuidutils
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy import sql
from apmec.common import exceptions
from apmec.db.common_services import common_services_db_plugin
from apmec.db import db_base
from apmec.db.mem import mem_db
from apmec.db.meo import meo_db
from apmec.extensions import meo
from apmec import manager
from apmec.plugins.common import constants
VIM_ATTRIBUTES = ('id', 'type', 'tenant_id', 'name', 'description',
'placement_attr', 'shared', 'is_default',

View File

@ -1,384 +0,0 @@
# 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 ast
from datetime import datetime
from oslo_db.exception import DBDuplicateEntry
from oslo_log import log as logging
from oslo_utils import timeutils
from oslo_utils import uuidutils
from six import iteritems
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy import schema
from apmec.common import exceptions
from apmec.db.common_services import common_services_db_plugin
from apmec.db import db_base
from apmec.db import model_base
from apmec.db import models_v1
from apmec.db import types
from apmec.extensions import meo
from apmec.extensions.meo_plugins import edge_service
from apmec.plugins.common import constants
LOG = logging.getLogger(__name__)
_ACTIVE_UPDATE = (constants.ACTIVE, constants.PENDING_UPDATE)
_ACTIVE_UPDATE_ERROR_DEAD = (
constants.PENDING_CREATE, constants.ACTIVE, constants.PENDING_UPDATE,
constants.ERROR, constants.DEAD)
CREATE_STATES = (constants.PENDING_CREATE, constants.DEAD)
###########################################################################
# db tables
class MESD(model_base.BASE, models_v1.HasId, models_v1.HasTenant,
models_v1.Audit):
"""Represents MESD to create MES."""
__tablename__ = 'mesd'
# Descriptive name
name = sa.Column(sa.String(255), nullable=False)
description = sa.Column(sa.Text)
meads = sa.Column(types.Json, nullable=True)
# Mesd template source - onboarded
template_source = sa.Column(sa.String(255), server_default='onboarded')
# (key, value) pair to spin up
attributes = orm.relationship('MESDAttribute',
backref='mesd')
__table_args__ = (
schema.UniqueConstraint(
"tenant_id",
"name",
name="uniq_mesd0tenant_id0name"),
)
class MESDAttribute(model_base.BASE, models_v1.HasId):
"""Represents attributes necessary for creation of mes in (key, value) pair
"""
__tablename__ = 'mesd_attribute'
mesd_id = sa.Column(types.Uuid, sa.ForeignKey('mesd.id'),
nullable=False)
key = sa.Column(sa.String(255), nullable=False)
value = sa.Column(sa.TEXT(65535), nullable=True)
class MES(model_base.BASE, models_v1.HasId, models_v1.HasTenant,
models_v1.Audit):
"""Represents network services that deploys services.
"""
__tablename__ = 'mes'
mesd_id = sa.Column(types.Uuid, sa.ForeignKey('mesd.id'))
mesd = orm.relationship('MESD')
name = sa.Column(sa.String(255), nullable=False)
description = sa.Column(sa.Text, nullable=True)
# Dict of MEA details that network service launches
mea_ids = sa.Column(sa.TEXT(65535), nullable=True)
# Dict of mgmt urls that network servic launches
mgmt_urls = sa.Column(sa.TEXT(65535), nullable=True)
status = sa.Column(sa.String(64), nullable=False)
vim_id = sa.Column(types.Uuid, sa.ForeignKey('vims.id'), nullable=False)
error_reason = sa.Column(sa.Text, nullable=True)
__table_args__ = (
schema.UniqueConstraint(
"tenant_id",
"name",
name="uniq_mes0tenant_id0name"),
)
class MESPluginDb(edge_service.MESPluginBase, db_base.CommonDbMixin):
def __init__(self):
super(MESPluginDb, self).__init__()
self._cos_db_plg = common_services_db_plugin.CommonServicesPluginDb()
def _get_resource(self, context, model, id):
try:
return self._get_by_id(context, model, id)
except orm_exc.NoResultFound:
if issubclass(model, MESD):
raise edge_service.MESDNotFound(mesd_id=id)
if issubclass(model, MES):
raise edge_service.MESNotFound(mes_id=id)
else:
raise
def _get_mes_db(self, context, mes_id, current_statuses, new_status):
try:
mes_db = (
self._model_query(context, MES).
filter(MES.id == mes_id).
filter(MES.status.in_(current_statuses)).
with_lockmode('update').one())
except orm_exc.NoResultFound:
raise edge_service.MESNotFound(mes_id=mes_id)
mes_db.update({'status': new_status})
return mes_db
def _make_attributes_dict(self, attributes_db):
return dict((attr.key, attr.value) for attr in attributes_db)
def _make_mesd_dict(self, mesd, fields=None):
res = {
'attributes': self._make_attributes_dict(mesd['attributes']),
}
key_list = ('id', 'tenant_id', 'name', 'description',
'created_at', 'updated_at', 'meads', 'template_source')
res.update((key, mesd[key]) for key in key_list)
return self._fields(res, fields)
def _make_dev_attrs_dict(self, dev_attrs_db):
return dict((arg.key, arg.value) for arg in dev_attrs_db)
def _make_mes_dict(self, mes_db, fields=None):
LOG.debug('mes_db %s', mes_db)
res = {}
key_list = ('id', 'tenant_id', 'mesd_id', 'name', 'description',
'mea_ids', 'status', 'mgmt_urls', 'error_reason',
'vim_id', 'created_at', 'updated_at')
res.update((key, mes_db[key]) for key in key_list)
return self._fields(res, fields)
def create_mesd(self, context, mesd):
meads = mesd['meads']
mesd = mesd['mesd']
LOG.debug('mesd %s', mesd)
tenant_id = self._get_tenant_id_for_create(context, mesd)
template_source = mesd.get('template_source')
try:
with context.session.begin(subtransactions=True):
mesd_id = uuidutils.generate_uuid()
mesd_db = MESD(
id=mesd_id,
tenant_id=tenant_id,
name=mesd.get('name'),
meads=meads,
description=mesd.get('description'),
deleted_at=datetime.min,
template_source=template_source)
context.session.add(mesd_db)
for (key, value) in mesd.get('attributes', {}).items():
attribute_db = MESDAttribute(
id=uuidutils.generate_uuid(),
mesd_id=mesd_id,
key=key,
value=value)
context.session.add(attribute_db)
except DBDuplicateEntry as e:
raise exceptions.DuplicateEntity(
_type="mesd",
entry=e.columns)
LOG.debug('mesd_db %(mesd_db)s %(attributes)s ',
{'mesd_db': mesd_db,
'attributes': mesd_db.attributes})
mesd_dict = self._make_mesd_dict(mesd_db)
LOG.debug('mesd_dict %s', mesd_dict)
self._cos_db_plg.create_event(
context, res_id=mesd_dict['id'],
res_type=constants.RES_TYPE_MESD,
res_state=constants.RES_EVT_ONBOARDED,
evt_type=constants.RES_EVT_CREATE,
tstamp=mesd_dict[constants.RES_EVT_CREATED_FLD])
return mesd_dict
def delete_mesd(self,
context,
mesd_id,
soft_delete=True):
with context.session.begin(subtransactions=True):
mess_db = context.session.query(MES).filter_by(
mesd_id=mesd_id).first()
if mess_db is not None and mess_db.deleted_at is None:
raise meo.MESDInUse(mesd_id=mesd_id)
mesd_db = self._get_resource(context, MESD,
mesd_id)
if soft_delete:
mesd_db.update({'deleted_at': timeutils.utcnow()})
self._cos_db_plg.create_event(
context, res_id=mesd_db['id'],
res_type=constants.RES_TYPE_MESD,
res_state=constants.RES_EVT_NA_STATE,
evt_type=constants.RES_EVT_DELETE,
tstamp=mesd_db[constants.RES_EVT_DELETED_FLD])
else:
context.session.query(MESDAttribute).filter_by(
mesd_id=mesd_id).delete()
context.session.delete(mesd_db)
def get_mesd(self, context, mesd_id, fields=None):
mesd_db = self._get_resource(context, MESD, mesd_id)
return self._make_mesd_dict(mesd_db)
def get_mesds(self, context, filters, fields=None):
if ('template_source' in filters) and \
(filters['template_source'][0] == 'all'):
filters.pop('template_source')
return self._get_collection(context, MESD,
self._make_mesd_dict,
filters=filters, fields=fields)
# reference implementation. needs to be overrided by subclass
def create_mes(self, context, mes):
LOG.debug('mes %s', mes)
mes = mes['mes']
tenant_id = self._get_tenant_id_for_create(context, mes)
mesd_id = mes['mesd_id']
vim_id = mes['vim_id']
name = mes.get('name')
mes_id = uuidutils.generate_uuid()
try:
with context.session.begin(subtransactions=True):
mesd_db = self._get_resource(context, MESD,
mesd_id)
mes_db = MES(id=mes_id,
tenant_id=tenant_id,
name=name,
description=mesd_db.description,
mea_ids=None,
status=constants.PENDING_CREATE,
mgmt_urls=None,
mesd_id=mesd_id,
vim_id=vim_id,
error_reason=None,
deleted_at=datetime.min)
context.session.add(mes_db)
except DBDuplicateEntry as e:
raise exceptions.DuplicateEntity(
_type="mes",
entry=e.columns)
evt_details = "MES UUID assigned."
self._cos_db_plg.create_event(
context, res_id=mes_id,
res_type=constants.RES_TYPE_mes,
res_state=constants.PENDING_CREATE,
evt_type=constants.RES_EVT_CREATE,
tstamp=mes_db[constants.RES_EVT_CREATED_FLD],
details=evt_details)
return self._make_mes_dict(mes_db)
def create_mes_post(self, context, mes_id, mistral_obj,
mead_dict, error_reason):
LOG.debug('mes ID %s', mes_id)
output = ast.literal_eval(mistral_obj.output)
mgmt_urls = dict()
mea_ids = dict()
if len(output) > 0:
for mead_name, mead_val in iteritems(mead_dict):
for instance in mead_val['instances']:
if 'mgmt_url_' + instance in output:
mgmt_urls[instance] = ast.literal_eval(
output['mgmt_url_' + instance].strip())
mea_ids[instance] = output['mea_id_' + instance]
mea_ids = str(mea_ids)
mgmt_urls = str(mgmt_urls)
if not mea_ids:
mea_ids = None
if not mgmt_urls:
mgmt_urls = None
status = constants.ACTIVE if mistral_obj.state == 'SUCCESS' \
else constants.ERROR
with context.session.begin(subtransactions=True):
mes_db = self._get_resource(context, MES,
mes_id)
mes_db.update({'mea_ids': mea_ids})
mes_db.update({'mgmt_urls': mgmt_urls})
mes_db.update({'status': status})
mes_db.update({'error_reason': error_reason})
mes_db.update({'updated_at': timeutils.utcnow()})
mes_dict = self._make_mes_dict(mes_db)
self._cos_db_plg.create_event(
context, res_id=mes_dict['id'],
res_type=constants.RES_TYPE_mes,
res_state=constants.RES_EVT_NA_STATE,
evt_type=constants.RES_EVT_UPDATE,
tstamp=mes_dict[constants.RES_EVT_UPDATED_FLD])
return mes_dict
# reference implementation. needs to be overrided by subclass
def delete_mes(self, context, mes_id):
with context.session.begin(subtransactions=True):
mes_db = self._get_mes_db(
context, mes_id, _ACTIVE_UPDATE_ERROR_DEAD,
constants.PENDING_DELETE)
deleted_mes_db = self._make_mes_dict(mes_db)
self._cos_db_plg.create_event(
context, res_id=mes_id,
res_type=constants.RES_TYPE_mes,
res_state=deleted_mes_db['status'],
evt_type=constants.RES_EVT_DELETE,
tstamp=timeutils.utcnow(), details="MES delete initiated")
return deleted_mes_db
def delete_mes_post(self, context, mes_id, mistral_obj,
error_reason, soft_delete=True):
mes = self.get_mes(context, mes_id)
mesd_id = mes.get('mesd_id')
with context.session.begin(subtransactions=True):
query = (
self._model_query(context, MES).
filter(MES.id == mes_id).
filter(MES.status == constants.PENDING_DELETE))
if mistral_obj and mistral_obj.state == 'ERROR':
query.update({'status': constants.ERROR})
self._cos_db_plg.create_event(
context, res_id=mes_id,
res_type=constants.RES_TYPE_mes,
res_state=constants.ERROR,
evt_type=constants.RES_EVT_DELETE,
tstamp=timeutils.utcnow(),
details="MES Delete ERROR")
else:
if soft_delete:
deleted_time_stamp = timeutils.utcnow()
query.update({'deleted_at': deleted_time_stamp})
self._cos_db_plg.create_event(
context, res_id=mes_id,
res_type=constants.RES_TYPE_mes,
res_state=constants.PENDING_DELETE,
evt_type=constants.RES_EVT_DELETE,
tstamp=deleted_time_stamp,
details="mes Delete Complete")
else:
query.delete()
template_db = self._get_resource(context, MESD, mesd_id)
if template_db.get('template_source') == 'inline':
self.delete_mesd(context, mesd_id)
def get_mes(self, context, mes_id, fields=None):
mes_db = self._get_resource(context, MES, mes_id)
return self._make_mes_dict(mes_db)
def get_mess(self, context, filters=None, fields=None):
return self._get_collection(context, MES,
self._make_mes_dict,
filters=filters, fields=fields)

View File

@ -40,7 +40,6 @@ def upgrade(active_plugins=None, options=None):
sa.Column('deleted_at', sa.DateTime(), nullable=True),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('meads', types.Json, nullable=True),
sa.PrimaryKeyConstraint('id'),
mysql_engine='InnoDB'
)

View File

@ -0,0 +1,44 @@
# Copyright 2018 OpenStack Foundation
#
# 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.
#
"""mesd-table
Revision ID: 20f6a08b066e
Revises: 8206737b5c80
Create Date: 2018-06-13 20:36:59.470322
"""
# revision identifiers, used by Alembic.
revision = '20f6a08b066e'
down_revision = '8206737b5c80'
from alembic import op
import sqlalchemy as sa
from apmec.db import types
def upgrade(active_plugins=None, options=None):
op.add_column('mesd',
sa.Column('mesd_mapping',
types.Json, nullable=True))
op.add_column('mes',
sa.Column('mes_mapping',
types.Json, nullable=True))
op.add_column('mes',
sa.Column('reused',
types.Json, nullable=True))

View File

@ -0,0 +1,81 @@
# Copyright 2018 OpenStack Foundation
#
# 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.
#
"""meca-table
Revision ID: 8206737b5c80
Revises: e9a1e47fb0b5
Create Date: 2018-06-08 15:58:43.286238
"""
# revision identifiers, used by Alembic.
revision = '8206737b5c80'
down_revision = 'e9a1e47fb0b5'
from alembic import op
import sqlalchemy as sa
from apmec.db import types
def upgrade(active_plugins=None, options=None):
op.create_table('mecad',
sa.Column('tenant_id',
sa.String(length=64), nullable=False),
sa.Column('id', types.Uuid(length=36), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True),
sa.Column('deleted_at', sa.DateTime(), nullable=True),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('meads', types.Json, nullable=True),
sa.Column('template_source', sa.String(length=255),
server_default='onboarded'),
sa.PrimaryKeyConstraint('id'),
mysql_engine='InnoDB'
)
op.create_table('meca',
sa.Column('tenant_id',
sa.String(length=64), nullable=False),
sa.Column('id', types.Uuid(length=36), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True),
sa.Column('deleted_at', sa.DateTime(), nullable=True),
sa.Column('mecad_id',
types.Uuid(length=36), nullable=True),
sa.Column('vim_id', sa.String(length=64), nullable=False),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('mea_ids', sa.TEXT(length=65535), nullable=True),
sa.Column('mgmt_urls',
sa.TEXT(length=65535), nullable=True),
sa.Column('status', sa.String(length=64), nullable=False),
sa.Column('error_reason', sa.Text(), nullable=True),
sa.ForeignKeyConstraint(['mecad_id'], ['mecad.id'], ),
sa.PrimaryKeyConstraint('id'),
mysql_engine='InnoDB'
)
op.create_table('mecad_attribute',
sa.Column('id', types.Uuid(length=36), nullable=False),
sa.Column('mecad_id',
types.Uuid(length=36), nullable=False),
sa.Column('key', sa.String(length=255), nullable=False),
sa.Column('value', sa.TEXT(length=65535), nullable=True),
sa.ForeignKeyConstraint(['mecad_id'], ['mecad.id'], ),
sa.PrimaryKeyConstraint('id'),
mysql_engine='InnoDB'
)

View File

@ -1 +1 @@
e9a1e47fb0b5
20f6a08b066e

View File

@ -86,116 +86,12 @@ class ToscaParserFailed(exceptions.InvalidInput):
message = _("tosca-parser failed: - %(error_msg_details)s")
class NfydInvalidTemplate(exceptions.InvalidInput):
message = _("Invalid NFY template input: %(template)s")
class MECADInUse(exceptions.InUse):
message = _('MECAD %(mecad_id)s is still in use')
class NfydDuplicateForwarderException(exceptions.InvalidInput):
message = _("Invalid Forwarding Path contains duplicate forwarder not in "
"order: %(forwarder)s")
class NfydDuplicateCPException(exceptions.InvalidInput):
message = _("Invalid Forwarding Path contains duplicate connection point "
": %(cp)s")
class NfydCpNotFoundException(exceptions.NotFound):
message = _("Specified CP %(cp_id)s could not be found in MEAD "
"%(mead_name)s. Please check MEAD for correct Connection "
"Point.")
class NfydCpNoForwardingException(exceptions.ApmecException):
message = _("Specified CP %(cp_id)s in MEAD %(mead_name)s "
"does not have forwarding capability, which is required to be "
"included in forwarding path")
class NfydWrongEndpointNumber(exceptions.ApmecException):
message = _("Specified number_of_endpoints %(number)s is not equal to "
"the number of connection_point %(cps)s")
class NfyInvalidMappingException(exceptions.ApmecException):
message = _("Matching MEA Instance for MEAD %(mead_name)s could not be "
"found. Please create an instance of this MEAD before "
"creating/updating NFY.")
class NfyParamValueFormatError(exceptions.ApmecException):
message = _("Param values %(param_value)s is not in dict format.")
class NfyParamValueNotUsed(exceptions.ApmecException):
message = _("Param input %(param_key)s not used.")
class NfyCpNotFoundException(exceptions.NotFound):
message = _("Specified CP %(cp_id)s could not be found in MEA "
"%(mea_id)s.")
class NfyMeaNotFoundException(exceptions.NotFound):
message = _("Specified MEA instance %(mea_name)s in MEA Mapping could not "
"be found")
class NfpAttributeNotFoundException(exceptions.NotFound):
message = _('NFP attribute %(attribute)s could not be found')
class NfpNotFoundException(exceptions.NotFound):
message = _('NFP %(nfp_id)s could not be found')
class NfpInUse(exceptions.InUse):
message = _('NFP %(nfp_id)s is still in use')
class NfpPolicyCriteriaError(exceptions.PolicyCheckError):
message = _('%(error)s in policy')
class NfpPolicyNotFoundException(exceptions.NotFound):
message = _('Policy not found in NFP %(nfp)s')
class NfpPolicyTypeError(exceptions.PolicyCheckError):
message = _('Unsupported Policy Type: %(type)s')
class NfpForwarderNotFoundException(exceptions.NotFound):
message = _('MEAD Forwarder %(mead)s not found in MEA Mapping %(mapping)s')
class NfpRequirementsException(exceptions.ApmecException):
message = _('MEAD Forwarder %(mead)s specified more than twice in '
'requirements path')
class SfcInUse(exceptions.InUse):
message = _('SFC %(sfc_id)s is still in use')
class SfcNotFoundException(exceptions.NotFound):
message = _('Service Function Chain %(sfc_id)s could not be found')
class ClassifierInUse(exceptions.InUse):
message = _('Classifier %(classifier_id)s is still in use')
class ClassifierNotFoundException(exceptions.NotFound):
message = _('Classifier %(classifier_id)s could not be found')
class MESDInUse(exceptions.InUse):
message = _('MESD %(mesd_id)s is still in use')
class MESInUse(exceptions.InUse):
message = _('MES %(mes_id)s is still in use')
class MECAInUse(exceptions.InUse):
message = _('MECA %(meca_id)s is still in use')
class NoTasksException(exceptions.ApmecException):
@ -292,7 +188,8 @@ RESOURCE_ATTRIBUTE_MAP = {
'is_visible': True,
},
},
'mesds': {
'mecads': {
'id': {
'allow_post': False,
'allow_put': False,
@ -347,7 +244,7 @@ RESOURCE_ATTRIBUTE_MAP = {
},
'mess': {
'mecas': {
'id': {
'allow_post': False,
'allow_put': False,
@ -392,7 +289,7 @@ RESOURCE_ATTRIBUTE_MAP = {
'is_visible': True,
'default': '',
},
'mesd_id': {
'mecad_id': {
'allow_post': True,
'allow_put': False,
'validate': {'type:uuid': None},
@ -431,15 +328,14 @@ RESOURCE_ATTRIBUTE_MAP = {
'validate': {'type:dict_or_nodata': None},
'is_visible': True,
},
'mesd_template': {
'mecad_template': {
'allow_post': True,
'allow_put': False,
'allow_put': True,
'validate': {'type:dict_or_nodata': None},
'is_visible': True,
'default': None,
},
},
}
@ -524,3 +420,51 @@ class MEOPluginBase(service_base.MECPluginBase):
def get_default_vim(self, context):
raise NotImplementedError()
@six.add_metaclass(abc.ABCMeta)
class MECAPluginBase(service_base.MECPluginBase):
@abc.abstractmethod
def create_mecad(self, context, mecad):
pass
@abc.abstractmethod
def delete_mecad(self, context, mecad_id):
pass
@abc.abstractmethod
def get_mecad(self, context, mecad_id, fields=None):
pass
@abc.abstractmethod
def get_mecads(self, context, filters=None, fields=None):
pass
@abc.abstractmethod
def create_meca(self, context, meca):
pass
@abc.abstractmethod
def get_mecas(self, context, filters=None, fields=None):
pass
@abc.abstractmethod
def get_meca(self, context, meca_id, fields=None):
pass
@abc.abstractmethod
def delete_meca(self, context, meca_id):
pass
@abc.abstractmethod
def update_meca(self, context, meca_id, meca):
pass
class MECADNotFound(exceptions.NotFound):
message = _('MECAD %(mecad_id)s could not be found')
class MECANotFound(exceptions.NotFound):
message = _('MECA %(meca_id)s could not be found')

View File

@ -19,23 +19,8 @@ import os
import time
import yaml
from apmec._i18n import _
from apmec.common import driver_manager
from apmec.common import log
from apmec.common import utils
from apmec.db.meo import meo_db_plugin
from apmec.db.meo import mes_db
from apmec.extensions import common_services as cs
from apmec.extensions import meo
from apmec.keymgr import API as KEYMGR_API
from apmec import manager
from apmec.mem import vim_client
from apmec.meo.workflows.vim_monitor import vim_monitor_utils
from apmec.catalogs.tosca import utils as toscautils
from cryptography import fernet
import eventlet
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import excutils
@ -44,8 +29,21 @@ from oslo_utils import uuidutils
from tempfile import mkstemp
from toscaparser.tosca_template import ToscaTemplate
from apmec._i18n import _
from apmec.common import driver_manager
from apmec.common import log
from apmec.common import utils
from apmec.db.meo import meca_db
from apmec.db.meo import meo_db_plugin
from apmec.extensions import common_services as cs
from apmec.extensions import meo
from apmec.keymgr import API as KEYMGR_API
from apmec import manager
from apmec.mem import vim_client
from apmec.meo.workflows.vim_monitor import vim_monitor_utils
from apmec.plugins.common import constants
from toscaparser import tosca_template
from apmec.catalogs.tosca import utils as toscautils
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
@ -57,7 +55,7 @@ def config_opts():
return [('meo_vim', MeoPlugin.OPTS)]
class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
class MeoPlugin(meo_db_plugin.MeoPluginDb, meca_db.MECAPluginDb):
"""MEO reference plugin for MEO extension
Implements the MEO extension and defines public facing APIs for VIM
@ -202,25 +200,6 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
auth_dict = self.get_auth_dict(context)
vim_monitor_utils.monitor_vim(auth_dict, vim_obj)
@log.log
def validate_tosca(self, template):
if "tosca_definitions_version" not in template:
raise meo.ToscaParserFailed(
error_msg_details='tosca_definitions_version missing in '
'template'
)
LOG.debug('template yaml: %s', template)
toscautils.updateimports(template)
try:
tosca_template.ToscaTemplate(
a_file=False, yaml_dict_tpl=template)
except Exception as e:
LOG.exception("tosca-parser error: %s", str(e))
raise meo.ToscaParserFailed(error_msg_details=str(e))
def _get_vim_from_mea(self, context, mea_id):
"""Figures out VIM based on a MEA
@ -301,34 +280,36 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
resource_name=name)
@log.log
def create_mesd(self, context, mesd):
mesd_data = mesd['mesd']
template = mesd_data['attributes'].get('mesd')
def create_mecad(self, context, mecad):
mecad_data = mecad['mecad']
template = mecad_data['attributes'].get('mecad')
if isinstance(template, dict):
mesd_data['attributes']['mesd'] = yaml.safe_dump(
mecad_data['attributes']['mecad'] = yaml.safe_dump(
template)
LOG.debug('mesd %s', mesd_data)
LOG.debug('mecad %s', mecad_data)
if 'template_source' in mesd_data:
template_source = mesd_data.get('template_source')
if 'template_source' in mecad_data:
template_source = mecad_data.get('template_source')
else:
template_source = "onboarded"
mesd['mesd']['template_source'] = template_source
mecad['mecad']['template_source'] = template_source
self._parse_template_input(context, mesd)
return super(MeoPlugin, self).create_mesd(
context, mesd)
self._parse_template_input(context, mecad)
return super(MeoPlugin, self).create_mecad(
context, mecad)
def _parse_template_input(self, context, mecad):
mecad_dict = mecad['mecad']
mecad_yaml = mecad_dict['attributes'].get('mecad')
inner_mecad_dict = yaml.safe_load(mecad_yaml)
mecad['meads'] = dict()
LOG.debug('mecad_dict: %s', inner_mecad_dict)
# From import we can deploy both NS and MEC Application
def _parse_template_input(self, context, mesd):
mesd_dict = mesd['mesd']
mesd_yaml = mesd_dict['attributes'].get('mesd')
inner_mesd_dict = yaml.safe_load(mesd_yaml)
mesd['meads'] = dict()
LOG.debug('mesd_dict: %s', inner_mesd_dict)
# Deploy MEC applications
mem_plugin = manager.ApmecManager.get_service_plugins()['MEM']
mead_imports = inner_mesd_dict['imports']['meads']
inner_mesd_dict['imports'] = []
mead_imports = inner_mecad_dict['imports']
inner_mecad_dict['imports'] = []
new_files = []
for mead_name in mead_imports:
mead = mem_plugin.get_mead(context, mead_name)
@ -336,7 +317,7 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
sm_dict = yaml.safe_load(mead['attributes']['mead'])[
'topology_template'][
'substitution_mappings']
mesd['meads'][sm_dict['node_type']] = mead['name']
mecad['meads'][sm_dict['node_type']] = mead['name']
# Ugly Hack to validate the child templates
# TODO(tbh): add support in tosca-parser to pass child
# templates as dict
@ -345,33 +326,33 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
fp.write(mead['attributes']['mead'])
os.close(fd)
new_files.append(temp_path)
inner_mesd_dict['imports'].append(temp_path)
inner_mecad_dict['imports'].append(temp_path)
# Prepend the apmec_defs.yaml import file with the full
# path to the file
toscautils.updateimports(inner_mesd_dict)
toscautils.updateimports(inner_mecad_dict)
try:
ToscaTemplate(a_file=False,
yaml_dict_tpl=inner_mesd_dict)
yaml_dict_tpl=inner_mecad_dict)
except Exception as e:
LOG.exception("tosca-parser error: %s", str(e))
raise meo.ToscaParserFailed(error_msg_details=str(e))
finally:
for file_path in new_files:
os.remove(file_path)
inner_mesd_dict['imports'] = mead_imports
inner_mecad_dict['imports'] = mead_imports
if ('description' not in mesd_dict or
mesd_dict['description'] == ''):
mesd_dict['description'] = inner_mesd_dict.get(
if ('description' not in mecad_dict or
mecad_dict['description'] == ''):
mecad_dict['description'] = inner_mecad_dict.get(
'description', '')
if (('name' not in mesd_dict or
not len(mesd_dict['name'])) and
'metadata' in inner_mesd_dict):
mesd_dict['name'] = inner_mesd_dict['metadata'].get(
if (('name' not in mecad_dict or
not len(mecad_dict['name'])) and
'metadata' in inner_mecad_dict):
mecad_dict['name'] = inner_mecad_dict['metadata'].get(
'template_name', '')
LOG.debug('mesd %s', mesd)
LOG.debug('mecad %s', mecad)
def _get_mead_id(self, mead_name, onboarded_meads):
for mead in onboarded_meads:
@ -379,52 +360,52 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
return mead['id']
@log.log
def create_mes(self, context, mes):
"""Create MES and corresponding MEAs.
def create_meca(self, context, meca):
"""Create MECA and corresponding MEAs.
:param mes: mes dict which contains mesd_id and attributes
:param meca: meca dict which contains mecad_id and attributes
This method has 3 steps:
step-1: substitute all get_input params to its corresponding values
step-2: Build params dict for substitution mappings case through which
MEAs will actually substitute their requirements.
step-3: Create mistral workflow and execute the workflow
"""
mes_info = mes['mes']
name = mes_info['name']
meca_info = meca['meca']
name = meca_info['name']
if mes_info.get('mesd_template'):
mesd_name = utils.generate_resource_name(name, 'inline')
mesd = {'mesd': {
'attributes': {'mesd': mes_info['mesd_template']},
'description': mes_info['description'],
'name': mesd_name,
if meca_info.get('mecad_template'):
mecad_name = utils.generate_resource_name(name, 'inline')
mecad = {'mecad': {
'attributes': {'mecad': meca_info['mecad_template']},
'description': meca_info['description'],
'name': mecad_name,
'template_source': 'inline',
'tenant_id': mes_info['tenant_id']}}
mes_info['mesd_id'] = self.create_mesd(context, mesd).get('id')
'tenant_id': meca_info['tenant_id']}}
meca_info['mecad_id'] = self.create_mecad(context, mecad).get('id')
mesd = self.get_mesd(context, mes['mes']['mesd_id'])
mesd_dict = yaml.safe_load(mesd['attributes']['mesd'])
mecad = self.get_mecad(context, meca['meca']['mecad_id'])
mecad_dict = yaml.safe_load(mecad['attributes']['mecad'])
mem_plugin = manager.ApmecManager.get_service_plugins()['MEM']
onboarded_meads = mem_plugin.get_meads(context, [])
region_name = mes.setdefault('placement_attr', {}).get(
region_name = meca.setdefault('placement_attr', {}).get(
'region_name', None)
vim_res = self.vim_client.get_vim(context, mes['mes']['vim_id'],
vim_res = self.vim_client.get_vim(context, meca['meca']['vim_id'],
region_name)
driver_type = vim_res['vim_type']
if not mes['mes']['vim_id']:
mes['mes']['vim_id'] = vim_res['vim_id']
if not meca['meca'].get('vim_id'):
meca['meca']['vim_id'] = vim_res['vim_id']
# Step-1
param_values = mes['mes']['attributes'].get('param_values', {})
if 'get_input' in str(mesd_dict):
self._process_parameterized_input(mes['mes']['attributes'],
mesd_dict)
param_values = meca['meca']['attributes'].get('param_values', {})
if 'get_input' in str(mecad_dict):
self._process_parameterized_input(meca['meca']['attributes'],
mecad_dict)
# Step-2
meads = mesd['meads']
meads = mecad['meads']
# mead_dict is used while generating workflow
mead_dict = dict()
for node_name, node_val in \
(mesd_dict['topology_template']['node_templates']).items():
(mecad_dict['topology_template']['node_templates']).items():
if node_val.get('type') not in meads.keys():
continue
mead_name = meads[node_val.get('type')]
@ -445,18 +426,18 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
for requirement in requirements:
req_name = list(requirement.keys())[0]
req_val = list(requirement.values())[0]
res_name = req_val + mes['mes']['mesd_id'][:11]
res_name = req_val + meca['meca']['mecad_id'][:11]
req_dict[req_name] = res_name
if req_val in mesd_dict['topology_template']['node_templates']:
if req_val in mecad_dict['topology_template']['node_templates']: # noqa
param_values[mead_name]['substitution_mappings'][
res_name] = mesd_dict['topology_template'][
res_name] = mecad_dict['topology_template'][
'node_templates'][req_val]
param_values[mead_name]['substitution_mappings'][
'requirements'] = req_dict
mes['mead_details'] = mead_dict
meca['mead_details'] = mead_dict
# Step-3
kwargs = {'mes': mes, 'params': param_values}
kwargs = {'meca': meca, 'params': param_values}
# NOTE NoTasksException is raised if no tasks.
workflow = self._vim_drivers.invoke(
@ -479,9 +460,9 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
workflow_id=workflow['id'],
auth_dict=self.get_auth_dict(context))
raise ex
mes_dict = super(MeoPlugin, self).create_mes(context, mes)
meca_dict = super(MeoPlugin, self).create_meca(context, meca)
def _create_mes_wait(self_obj, mes_id, execution_id):
def _create_meca_wait(self_obj, meca_id, execution_id):
exec_state = "RUNNING"
mistral_retries = MISTRAL_RETRIES
while exec_state == "RUNNING" and mistral_retries > 0:
@ -498,7 +479,7 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
error_reason = None
if mistral_retries == 0 and exec_state == 'RUNNING':
error_reason = _(
"MES creation is not completed within"
"MECA creation is not completed within"
" {wait} seconds as creation of mistral"
" execution {mistral} is not completed").format(
wait=MISTRAL_RETRIES * MISTRAL_RETRY_WAIT,
@ -516,12 +497,12 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
'delete_workflow',
workflow_id=workflow['id'],
auth_dict=self.get_auth_dict(context))
super(MeoPlugin, self).create_mes_post(context, mes_id, exec_obj,
super(MeoPlugin, self).create_meca_post(context, meca_id, exec_obj,
mead_dict, error_reason)
self.spawn_n(_create_mes_wait, self, mes_dict['id'],
self.spawn_n(_create_meca_wait, self, meca_dict['id'],
mistral_execution.id)
return mes_dict
return meca_dict
@log.log
def _update_params(self, original, paramvalues):
@ -539,20 +520,20 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
self._update_params(value, paramvalues)
@log.log
def _process_parameterized_input(self, attrs, mesd_dict):
def _process_parameterized_input(self, attrs, mecad_dict):
param_vattrs_dict = attrs.pop('param_values', None)
if param_vattrs_dict:
for node in \
mesd_dict['topology_template']['node_templates'].values():
mecad_dict['topology_template']['node_templates'].values():
if 'get_input' in str(node):
self._update_params(node, param_vattrs_dict['mesd'])
self._update_params(node, param_vattrs_dict['mecad'])
else:
raise cs.ParamYAMLInputMissing()
@log.log
def delete_mes(self, context, mes_id):
mes = super(MeoPlugin, self).get_mes(context, mes_id)
vim_res = self.vim_client.get_vim(context, mes['vim_id'])
def delete_meca(self, context, meca_id):
meca = super(MeoPlugin, self).get_meca(context, meca_id)
vim_res = self.vim_client.get_vim(context, meca['vim_id'])
driver_type = vim_res['vim_type']
workflow = None
try:
@ -563,7 +544,7 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
action='delete',
auth_dict=self.get_auth_dict(context),
kwargs={
'mes': mes})
'meca': meca})
except meo.NoTasksException:
LOG.warning("No MEA deletion task(s).")
if workflow:
@ -582,9 +563,9 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
auth_dict=self.get_auth_dict(context))
raise ex
super(MeoPlugin, self).delete_mes(context, mes_id)
super(MeoPlugin, self).delete_meca(context, meca_id)
def _delete_mes_wait(mes_id, execution_id):
def _delete_meca_wait(meca_id, execution_id):
exec_state = "RUNNING"
mistral_retries = MISTRAL_RETRIES
while exec_state == "RUNNING" and mistral_retries > 0:
@ -601,7 +582,7 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
error_reason = None
if mistral_retries == 0 and exec_state == 'RUNNING':
error_reason = _(
"MES deletion is not completed within"
"MECA deletion is not completed within"
" {wait} seconds as deletion of mistral"
" execution {mistral} is not completed").format(
wait=MISTRAL_RETRIES * MISTRAL_RETRY_WAIT,
@ -619,11 +600,154 @@ class MeoPlugin(meo_db_plugin.MeoPluginDb, mes_db.MESPluginDb):
'delete_workflow',
workflow_id=workflow['id'],
auth_dict=self.get_auth_dict(context))
super(MeoPlugin, self).delete_mes_post(context, mes_id, exec_obj,
super(MeoPlugin, self).delete_meca_post(context, meca_id, exec_obj,
error_reason)
if workflow:
self.spawn_n(_delete_mes_wait, mes['id'], mistral_execution.id)
self.spawn_n(_delete_meca_wait, meca['id'], mistral_execution.id)
else:
super(MeoPlugin, self).delete_mes_post(
context, mes_id, None, None)
return mes['id']
super(MeoPlugin, self).delete_meca_post(
context, meca_id, None, None)
return meca['id']
@log.log
def update_meca(self, context, meca_id, meca):
meca_info = meca['meca']
meca_old = super(MeoPlugin, self).get_meca(context, meca_id)
name = meca_old['name']
# create inline meafgd if given by user
if meca_info.get('mecad_template'):
meca_name = utils.generate_resource_name(name, 'inline')
mecad = {'mecad': {'tenant_id': meca_old['tenant_id'],
'name': meca_name,
'attributes': {
'mecad': meca_info['mecad_template']},
'template_source': 'inline',
'description': meca_old['description']}}
try:
meca_info['mecad_id'] = \
self.create_mecad(context, mecad).get('id')
except Exception:
with excutils.save_and_reraise_exception():
super(MeoPlugin, self)._update_meca_status(
context, meca_id, constants.ACTIVE)
mecad = self.get_mecad(context, meca_info['mecad_id'])
mecad_dict = yaml.safe_load(mecad['attributes']['mecad'])
mem_plugin = manager.ApmecManager.get_service_plugins()['MEM']
onboarded_meads = mem_plugin.get_meads(context, [])
region_name = meca.setdefault('placement_attr', {}).get(
'region_name', None)
vim_res = self.vim_client.get_vim(context, meca_old['vim_id'],
region_name)
driver_type = vim_res['vim_type']
# Step-1
param_values = dict()
if 'get_input' in str(mecad_dict):
self._process_parameterized_input(meca['meca']['attributes'],
mecad_dict)
# Step-2
meads = mecad['meads']
# mead_dict is used while generating workflow
mead_dict = dict()
for node_name, node_val in \
(mecad_dict['topology_template']['node_templates']).items():
if node_val.get('type') not in meads.keys():
continue
mead_name = meads[node_val.get('type')]
if not mead_dict.get(mead_name):
mead_dict[mead_name] = {
'id': self._get_mead_id(mead_name, onboarded_meads),
'instances': [node_name]
}
else:
mead_dict[mead_name]['instances'].append(node_name)
if not node_val.get('requirements'):
continue
if not param_values.get(mead_name):
param_values[mead_name] = {}
param_values[mead_name]['substitution_mappings'] = dict()
req_dict = dict()
requirements = node_val.get('requirements')
for requirement in requirements:
req_name = list(requirement.keys())[0]
req_val = list(requirement.values())[0]
res_name = req_val + meca['meca']['mecad_id'][:11]
req_dict[req_name] = res_name
if req_val in mecad_dict['topology_template']['node_templates']: # noqa
param_values[mead_name]['substitution_mappings'][
res_name] = mecad_dict['topology_template'][
'node_templates'][req_val]
param_values[mead_name]['substitution_mappings'][
'requirements'] = req_dict
meca['mead_details'] = mead_dict
# Step-3
kwargs = {'meca': meca, 'params': param_values}
# NOTE NoTasksException is raised if no tasks.
workflow = self._vim_drivers.invoke(
driver_type,
'prepare_and_create_workflow',
resource='mea',
action='create',
auth_dict=self.get_auth_dict(context),
kwargs=kwargs)
try:
mistral_execution = self._vim_drivers.invoke(
driver_type,
'execute_workflow',
workflow=workflow,
auth_dict=self.get_auth_dict(context))
except Exception as ex:
LOG.error('Error while executing workflow: %s', ex)
self._vim_drivers.invoke(driver_type,
'delete_workflow',
workflow_id=workflow['id'],
auth_dict=self.get_auth_dict(context))
raise ex
meca_dict = super(MeoPlugin, self)._update_meca_pre(context, meca_id)
def _update_meca_wait(self_obj, meca_id, execution_id):
exec_state = "RUNNING"
mistral_retries = MISTRAL_RETRIES
while exec_state == "RUNNING" and mistral_retries > 0:
time.sleep(MISTRAL_RETRY_WAIT)
exec_state = self._vim_drivers.invoke(
driver_type,
'get_execution',
execution_id=execution_id,
auth_dict=self.get_auth_dict(context)).state
LOG.debug('status: %s', exec_state)
if exec_state == 'SUCCESS' or exec_state == 'ERROR':
break
mistral_retries = mistral_retries - 1
error_reason = None
if mistral_retries == 0 and exec_state == 'RUNNING':
error_reason = _(
"MECA update is not completed within"
" {wait} seconds as creation of mistral"
" execution {mistral} is not completed").format(
wait=MISTRAL_RETRIES * MISTRAL_RETRY_WAIT,
mistral=execution_id)
exec_obj = self._vim_drivers.invoke(
driver_type,
'get_execution',
execution_id=execution_id,
auth_dict=self.get_auth_dict(context))
self._vim_drivers.invoke(driver_type,
'delete_execution',
execution_id=execution_id,
auth_dict=self.get_auth_dict(context))
self._vim_drivers.invoke(driver_type,
'delete_workflow',
workflow_id=workflow['id'],
auth_dict=self.get_auth_dict(context))
super(MeoPlugin, self)._update_meca_post(
context, meca_id, exec_obj, mead_dict, error_reason)
self.spawn_n(_update_meca_wait, self, meca_dict['id'],
mistral_execution.id)
return meca_dict

View File

@ -37,9 +37,10 @@
apmec.db.common_services.common_services_db_plugin.rst
apmec.db.db_base.rst
apmec.db.mem.mem_db.rst
apmec.db.meo.meca_db.rst
apmec.db.meo.meo_db.rst
apmec.db.meo.meo_db_plugin.rst
apmec.db.meo.mes_db.rst
apmec.db.meso.meso_db.rst
apmec.db.migration.cli.rst
apmec.db.migration.models.head.rst
apmec.db.migration.purge_tables.rst
@ -50,7 +51,7 @@
apmec.extensions.common_services.rst
apmec.extensions.mem.rst
apmec.extensions.meo.rst
apmec.extensions.meo_plugins.edge_service.rst
apmec.extensions.meso.rst
apmec.hacking.checks.rst
apmec.keymgr.barbican_key_manager.rst
apmec.keymgr.exception.rst
@ -86,6 +87,9 @@
apmec.meo.workflows.vim_monitor.vim_monitor_utils.rst
apmec.meo.workflows.vim_monitor.vim_ping_action.rst
apmec.meo.workflows.vim_monitor.workflow_generator.rst
apmec.meso.drivers.nfv_drivers.abstract_driver.rst
apmec.meso.drivers.nfv_drivers.tacker_driver.rst
apmec.meso.meso_plugin.rst
apmec.mistral.actionrpc.kill_action.rst
apmec.mistral.mistral_client.rst
apmec.mistral.workflow_generator.rst