Commit 51ac8f60 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Context component: 

- extended grpc-to-enum tool to support arbitrary enum item names
- added PolicyRuleState enum model
- misc minor comment corrections
- misc import reorderings
- migrated Connection model and methods
- migrated PolicyRule model and methods
- removed unused files
parent f39c8d07

src/context/service/ContextServiceServicerImpl.py

+65 −172
Original line number Diff line number Diff line
@@ -12,10 +12,7 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.





import grpc, json, logging, sqlalchemy

#from sqlalchemy.orm import Session, contains_eager, selectinload, sessionmaker

#from sqlalchemy.dialects.postgresql import UUID, insert

from typing import Iterator

from common.message_broker.MessageBroker import MessageBroker

from common.proto.context_pb2 import (
@@ -27,45 +24,23 @@ from common.proto.context_pb2 import ( 
    Service, ServiceEvent, ServiceId, ServiceIdList, ServiceList,

    Slice, SliceEvent, SliceId, SliceIdList, SliceList,

    Topology, TopologyEvent, TopologyId, TopologyIdList, TopologyList)

#from common.proto.policy_pb2 import PolicyRuleIdList, PolicyRuleId, PolicyRuleList, PolicyRule

from common.proto.policy_pb2 import PolicyRuleIdList, PolicyRuleId, PolicyRuleList, PolicyRule

from common.proto.context_pb2_grpc import ContextServiceServicer

from common.proto.context_policy_pb2_grpc import ContextPolicyServiceServicer

#from common.tools.object_factory.Context import json_context_id

from common.rpc_method_wrapper.Decorator import create_metrics, safe_and_metered_rpc_method

#from common.rpc_method_wrapper.ServiceExceptions import (

#    InvalidArgumentException, NotFoundException, OperationFailedException)

from .database.Connection import (

    connection_delete, connection_get, connection_list_ids, connection_list_objs, connection_set)

from .database.Context import context_delete, context_get, context_list_ids, context_list_objs, context_set

from .database.Device import device_delete, device_get, device_list_ids, device_list_objs, device_set

from .database.Link import link_delete, link_get, link_list_ids, link_list_objs, link_set

from .database.PolicyRule import (

    policyrule_delete, policyrule_get, policyrule_list_ids, policyrule_list_objs, policyrule_set)

from .database.Service import service_delete, service_get, service_list_ids, service_list_objs, service_set

from .database.Slice import slice_delete, slice_get, slice_list_ids, slice_list_objs, slice_set, slice_unset

from .database.Topology import topology_delete, topology_get, topology_list_ids, topology_list_objs, topology_set

#from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string

#from context.service.Database import Database

#from context.service.database.ConfigModel import (

#    ConfigModel, ORM_ConfigActionEnum, ConfigRuleModel, grpc_config_rules_to_raw, update_config)

#from context.service.database.ConnectionModel import ConnectionModel, set_path

#from context.service.database.ConstraintModel import (

#    ConstraintModel, ConstraintsModel, Union_ConstraintModel, CONSTRAINT_PARSERS, set_constraints)

#from context.service.database.models.ContextModel import ContextModel

#from context.service.database.models.DeviceModel import (

#    DeviceModel, grpc_to_enum__device_operational_status, grpc_to_enum__device_driver)

#from context.service.database.models.EndPointModel import EndPointModel, grpc_to_enum__kpi_sample_type

#from context.service.database.EndPointModel import EndPointModel, set_kpi_sample_types

#from context.service.database.Events import notify_event

#from context.service.database.LinkModel import LinkModel

#from context.service.database.PolicyRuleModel import PolicyRuleModel

#from context.service.database.RelationModels import TopologyDeviceModel

#    ConnectionSubServiceModel, LinkEndPointModel, ServiceEndPointModel, SliceEndPointModel, SliceServiceModel,

#    SliceSubSliceModel, TopologyLinkModel)

#from context.service.database.ServiceModel import (

#    ServiceModel, grpc_to_enum__service_status, grpc_to_enum__service_type)

#from context.service.database.SliceModel import SliceModel, grpc_to_enum__slice_status

#from context.service.database.TopologyModel import TopologyModel

from .Constants import (

    CONSUME_TIMEOUT, TOPIC_CONNECTION, TOPIC_CONTEXT, TOPIC_DEVICE, TOPIC_LINK, #TOPIC_POLICY,

    TOPIC_SERVICE, TOPIC_SLICE, TOPIC_TOPOLOGY)

#from .ChangeFeedClient import ChangeFeedClient



LOGGER = logging.getLogger(__name__)
@@ -109,14 +84,14 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetContext(self, request : Context, context : grpc.ServicerContext) -> ContextId:

        context_id,updated = context_set(self.db_engine, request)

        context_id,updated = context_set(self.db_engine, request) # pylint: disable=unused-variable

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

        #notify_event(self.messagebroker, TOPIC_CONTEXT, event_type, {'context_id': context_id})

        return context_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveContext(self, request : ContextId, context : grpc.ServicerContext) -> Empty:

        deleted = context_delete(self.db_engine, request)

        deleted = context_delete(self.db_engine, request) # pylint: disable=unused-variable

        #if deleted:

        #    notify_event(self.messagebroker, TOPIC_CONTEXT, EventTypeEnum.EVENTTYPE_REMOVE, {'context_id': request})

        return Empty()
@@ -143,14 +118,14 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetTopology(self, request : Topology, context : grpc.ServicerContext) -> TopologyId:

        topology_id,updated = topology_set(self.db_engine, request)

        topology_id,updated = topology_set(self.db_engine, request) # pylint: disable=unused-variable

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

        #notify_event(self.messagebroker, TOPIC_TOPOLOGY, event_type, {'topology_id': topology_id})

        return topology_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveTopology(self, request : TopologyId, context : grpc.ServicerContext) -> Empty:

        deleted = topology_delete(self.db_engine, request)

        deleted = topology_delete(self.db_engine, request) # pylint: disable=unused-variable

        #if deleted:

        #    notify_event(self.messagebroker, TOPIC_TOPOLOGY, EventTypeEnum.EVENTTYPE_REMOVE, {'topology_id': request})

        return Empty()
@@ -177,14 +152,14 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetDevice(self, request : Device, context : grpc.ServicerContext) -> DeviceId:

        device_id,updated = device_set(self.db_engine, request)

        device_id,updated = device_set(self.db_engine, request) # pylint: disable=unused-variable

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

        #notify_event(self.messagebroker, TOPIC_DEVICE, event_type, {'device_id': device_id})

        return device_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveDevice(self, request : DeviceId, context : grpc.ServicerContext) -> Empty:

        deleted = device_delete(self.db_engine, request)

        deleted = device_delete(self.db_engine, request) # pylint: disable=unused-variable

        #if deleted:

        #    notify_event(self.messagebroker, TOPIC_DEVICE, EventTypeEnum.EVENTTYPE_REMOVE, {'device_id': request})

        return Empty()
@@ -211,14 +186,14 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetLink(self, request : Link, context : grpc.ServicerContext) -> LinkId:

        link_id,updated = link_set(self.db_engine, request)

        link_id,updated = link_set(self.db_engine, request) # pylint: disable=unused-variable

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

        #notify_event(self.messagebroker, TOPIC_LINK, event_type, {'link_id': link_id})

        return link_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveLink(self, request : LinkId, context : grpc.ServicerContext) -> Empty:

        deleted = link_delete(self.db_engine, request)

        deleted = link_delete(self.db_engine, request) # pylint: disable=unused-variable

        #if deleted:

        #    notify_event(self.messagebroker, TOPIC_LINK, EventTypeEnum.EVENTTYPE_REMOVE, {'link_id': request})

        return Empty()
@@ -245,14 +220,14 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetService(self, request : Service, context : grpc.ServicerContext) -> ServiceId:

        service_id,updated = service_set(self.db_engine, request)

        service_id,updated = service_set(self.db_engine, request) # pylint: disable=unused-variable

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

        #notify_event(self.messagebroker, TOPIC_SERVICE, event_type, {'service_id': service_id})

        return service_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveService(self, request : ServiceId, context : grpc.ServicerContext) -> Empty:

        deleted = service_delete(self.db_engine, request)

        deleted = service_delete(self.db_engine, request) # pylint: disable=unused-variable

        #if deleted:

        #    notify_event(self.messagebroker, TOPIC_SERVICE, EventTypeEnum.EVENTTYPE_REMOVE, {'service_id': request})

        return Empty()
@@ -279,21 +254,21 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetSlice(self, request : Slice, context : grpc.ServicerContext) -> SliceId:

        slice_id,updated = slice_set(self.db_engine, request)

        slice_id,updated = slice_set(self.db_engine, request) # pylint: disable=unused-variable

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

        #notify_event(self.messagebroker, TOPIC_SLICE, event_type, {'slice_id': slice_id})

        return slice_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def UnsetSlice(self, request : Slice, context : grpc.ServicerContext) -> SliceId:

        slice_id,updated = slice_unset(self.db_engine, request)

        slice_id,updated = slice_unset(self.db_engine, request) # pylint: disable=unused-variable

        #if updated:

        #    notify_event(self.messagebroker, TOPIC_SLICE, EventTypeEnum.EVENTTYPE_UPDATE, {'slice_id': slice_id})

        return slice_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveSlice(self, request : SliceId, context : grpc.ServicerContext) -> Empty:

        deleted = slice_delete(self.db_engine, request)

        deleted = slice_delete(self.db_engine, request) # pylint: disable=unused-variable

        #if deleted:

        #    notify_event(self.messagebroker, TOPIC_SLICE, EventTypeEnum.EVENTTYPE_REMOVE, {'slice_id': request})

        return Empty()
@@ -306,86 +281,32 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    # ----- Connection -------------------------------------------------------------------------------------------------



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def ListConnectionIds(self, request : ServiceId, context : grpc.ServicerContext) -> ConnectionIdList:

#        with self.session() as session:

#            result = session.query(DeviceModel).all()

#            return DeviceIdList(device_ids=[device.dump_id() for device in result])

#

#        with self.lock:

#            str_key = key_to_str([request.context_id.context_uuid.uuid, request.service_uuid.uuid])

#            db_service : ServiceModel = get_object(self.database, ServiceModel, str_key)

#            db_connections : Set[ConnectionModel] = get_related_objects(db_service, ConnectionModel)

#            db_connections = sorted(db_connections, key=operator.attrgetter('pk'))

#            return ConnectionIdList(connection_ids=[db_connection.dump_id() for db_connection in db_connections])



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def ListConnections(self, request : ContextId, context : grpc.ServicerContext) -> ServiceList:

#        with self.lock:

#            str_key = key_to_str([request.context_id.context_uuid.uuid, request.service_uuid.uuid])

#            db_service : ServiceModel = get_object(self.database, ServiceModel, str_key)

#            db_connections : Set[ConnectionModel] = get_related_objects(db_service, ConnectionModel)

#            db_connections = sorted(db_connections, key=operator.attrgetter('pk'))

#            return ConnectionList(connections=[db_connection.dump() for db_connection in db_connections])



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def GetConnection(self, request : ConnectionId, context : grpc.ServicerContext) -> Connection:

#        with self.lock:

#            db_connection : ConnectionModel = get_object(self.database, ConnectionModel, request.connection_uuid.uuid)

#            return Connection(**db_connection.dump(include_path=True, include_sub_service_ids=True))



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def SetConnection(self, request : Connection, context : grpc.ServicerContext) -> ConnectionId:

#        with self.lock:

#            connection_uuid = request.connection_id.connection_uuid.uuid

#

#            connection_attributes = {'connection_uuid': connection_uuid}

#

#            service_context_uuid = request.service_id.context_id.context_uuid.uuid

#            service_uuid = request.service_id.service_uuid.uuid

#            if len(service_context_uuid) > 0 and len(service_uuid) > 0:

#                str_service_key = key_to_str([service_context_uuid, service_uuid])

#                db_service : ServiceModel = get_object(self.database, ServiceModel, str_service_key)

#                connection_attributes['service_fk'] = db_service

#

#            path_hops_result = set_path(self.database, connection_uuid, request.path_hops_endpoint_ids, path_name = '')

#            db_path = path_hops_result[0]

#            connection_attributes['path_fk'] = db_path

#

#            result : Tuple[ConnectionModel, bool] = update_or_create_object(

#                self.database, ConnectionModel, connection_uuid, connection_attributes)

#            db_connection, updated = result

#

#            for sub_service_id in request.sub_service_ids:

#                sub_service_uuid         = sub_service_id.service_uuid.uuid

#                sub_service_context_uuid = sub_service_id.context_id.context_uuid.uuid

#                str_sub_service_key = key_to_str([sub_service_context_uuid, sub_service_uuid])

#                db_service : ServiceModel = get_object(self.database, ServiceModel, str_sub_service_key)

#

#                str_connection_sub_service_key = key_to_str([connection_uuid, str_sub_service_key], separator='--')

#                result : Tuple[ConnectionSubServiceModel, bool] = get_or_create_object(

#                    self.database, ConnectionSubServiceModel, str_connection_sub_service_key, {

#                        'connection_fk': db_connection, 'sub_service_fk': db_service})

#                #db_connection_sub_service, connection_sub_service_created = result

#

    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def ListConnectionIds(self, request : ServiceId, context : grpc.ServicerContext) -> ConnectionIdList:

        return connection_list_ids(self.db_engine, request)



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def ListConnections(self, request : ContextId, context : grpc.ServicerContext) -> ConnectionList:

        return connection_list_objs(self.db_engine, request)



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def GetConnection(self, request : ConnectionId, context : grpc.ServicerContext) -> Connection:

        return connection_get(self.db_engine, request)



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetConnection(self, request : Connection, context : grpc.ServicerContext) -> ConnectionId:

        connection_id,updated = connection_set(self.db_engine, request) # pylint: disable=unused-variable

        #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

#            dict_connection_id = db_connection.dump_id()

#            notify_event(self.messagebroker, TOPIC_CONNECTION, event_type, {'connection_id': dict_connection_id})

#            return ConnectionId(**dict_connection_id)



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def RemoveConnection(self, request : ConnectionId, context : grpc.ServicerContext) -> Empty:

#        with self.lock:

#            db_connection = ConnectionModel(self.database, request.connection_uuid.uuid, auto_load=False)

#            found = db_connection.load()

#            if not found: return Empty()

#

#            dict_connection_id = db_connection.dump_id()

#            db_connection.delete()

#

        #notify_event(self.messagebroker, TOPIC_CONNECTION, event_type, {'connection_id': connection_id})

        return connection_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemoveConnection(self, request : ConnectionId, context : grpc.ServicerContext) -> Empty:

        deleted = connection_delete(self.db_engine, request) # pylint: disable=unused-variable

        #if deleted:

        #    event_type = EventTypeEnum.EVENTTYPE_REMOVE

#            notify_event(self.messagebroker, TOPIC_CONNECTION, event_type, {'connection_id': dict_connection_id})

#            return Empty()

        #    notify_event(self.messagebroker, TOPIC_CONNECTION, event_type, {'connection_id': request})

        return Empty()



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def GetConnectionEvents(self, request : Empty, context : grpc.ServicerContext) -> Iterator[ConnectionEvent]:
@@ -395,52 +316,24 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    # ----- Policy -----------------------------------------------------------------------------------------------------



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def ListPolicyRuleIds(self, request : Empty, context: grpc.ServicerContext) -> PolicyRuleIdList:

#        with self.lock:

#            db_policy_rules: List[PolicyRuleModel] = get_all_objects(self.database, PolicyRuleModel)

#            db_policy_rules = sorted(db_policy_rules, key=operator.attrgetter('pk'))

#            return PolicyRuleIdList(policyRuleIdList=[db_policy_rule.dump_id() for db_policy_rule in db_policy_rules])



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def ListPolicyRules(self, request : Empty, context: grpc.ServicerContext) -> PolicyRuleList:

#        with self.lock:

#            db_policy_rules: List[PolicyRuleModel] = get_all_objects(self.database, PolicyRuleModel)

#            db_policy_rules = sorted(db_policy_rules, key=operator.attrgetter('pk'))

#            return PolicyRuleList(policyRules=[db_policy_rule.dump() for db_policy_rule in db_policy_rules])



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def GetPolicyRule(self, request : PolicyRuleId, context: grpc.ServicerContext) -> PolicyRule:

#        with self.lock:

#            policy_rule_uuid = request.uuid.uuid

#            db_policy_rule: PolicyRuleModel = get_object(self.database, PolicyRuleModel, policy_rule_uuid)

#            return PolicyRule(**db_policy_rule.dump())



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def SetPolicyRule(self, request : PolicyRule, context: grpc.ServicerContext) -> PolicyRuleId:

#        with self.lock:

#            policy_rule_type = request.WhichOneof('policy_rule')

#            policy_rule_json = grpc_message_to_json(request)

#            policy_rule_uuid = policy_rule_json[policy_rule_type]['policyRuleBasic']['policyRuleId']['uuid']['uuid']

#            result: Tuple[PolicyRuleModel, bool] = update_or_create_object(

#                self.database, PolicyRuleModel, policy_rule_uuid, {'value': json.dumps(policy_rule_json)})

#            db_policy, updated = result

#

#            #event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

#            dict_policy_id = db_policy.dump_id()

#            #notify_event(self.messagebroker, TOPIC_POLICY, event_type, {"policy_id": dict_policy_id})

#            return PolicyRuleId(**dict_policy_id)



#    @safe_and_metered_rpc_method(METRICS, LOGGER)

#    def RemovePolicyRule(self, request : PolicyRuleId, context: grpc.ServicerContext) -> Empty:

#        with self.lock:

#            policy_uuid = request.uuid.uuid

#            db_policy = PolicyRuleModel(self.database, policy_uuid, auto_load=False)

#            found = db_policy.load()

#            if not found: return Empty()

#

#            dict_policy_id = db_policy.dump_id()

#            db_policy.delete()

#            #event_type = EventTypeEnum.EVENTTYPE_REMOVE

#            #notify_event(self.messagebroker, TOPIC_POLICY, event_type, {"policy_id": dict_policy_id})

#            return Empty()
 
    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def ListPolicyRuleIds(self, request : Empty, context: grpc.ServicerContext) -> PolicyRuleIdList:

        return policyrule_list_ids(self.db_engine)



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def ListPolicyRules(self, request : Empty, context: grpc.ServicerContext) -> PolicyRuleList:

        return policyrule_list_objs(self.db_engine)



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def GetPolicyRule(self, request : PolicyRuleId, context: grpc.ServicerContext) -> PolicyRule:

        return policyrule_get(self.db_engine, request)



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def SetPolicyRule(self, request : PolicyRule, context: grpc.ServicerContext) -> PolicyRuleId:

        policyrule_id,updated = policyrule_set(self.db_engine, request) # pylint: disable=unused-variable

        return policyrule_id



    @safe_and_metered_rpc_method(METRICS, LOGGER)

    def RemovePolicyRule(self, request : PolicyRuleId, context: grpc.ServicerContext) -> Empty:

        deleted = policyrule_delete(self.db_engine, request) # pylint: disable=unused-variable

        return Empty()

src/context/service/Database.py

deleted100644 → 0
+0 −131

File deleted.

Preview size limit exceeded, changes collapsed.

src/context/service/database/Connection.py

0 → 100644
+134 −0

File added.

Preview size limit exceeded, changes collapsed.

src/context/service/database/PolicyRule.py

0 → 100644
+129 −0

File added.

Preview size limit exceeded, changes collapsed.

src/context/service/database/Slice.py

+1 −1
Original line number Diff line number Diff line
@@ -12,7 +12,7 @@ 
# See the License for the specific language governing permissions and

# limitations under the License.



from sqlalchemy import and_, delete

from sqlalchemy import and_

from sqlalchemy.dialects.postgresql import insert

from sqlalchemy.engine import Engine

from sqlalchemy.orm import Session, sessionmaker