Commit a3f7e8f7 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Context: 

- migrated events for Service and Slice entities
- added missing not-nulls to database fields
parent b9db50d5

src/context/service/ContextServiceServicerImpl.py

+27 −24
Original line number Diff line number Diff line
@@ -200,28 +200,29 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

    def ListServiceIds(self, request : ContextId, context : grpc.ServicerContext) -> ServiceIdList:

        return service_list_ids(self.db_engine, request)

        return ServiceIdList(service_ids=service_list_ids(self.db_engine, request))



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        return service_list_objs(self.db_engine, request)

        return ServiceList(services=service_list_objs(self.db_engine, request))



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        return service_get(self.db_engine, request)

        return Service(**service_get(self.db_engine, request))



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        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

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

        event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

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

        return ServiceId(**service_id)



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        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})

        service_id,deleted = service_delete(self.db_engine, request)

        if deleted:

            event_type = EventTypeEnum.EVENTTYPE_REMOVE

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

        return Empty()



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
@@ -234,35 +235,37 @@ class ContextServiceServicerImpl(ContextServiceServicer, ContextPolicyServiceSer 


    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

    def ListSliceIds(self, request : ContextId, context : grpc.ServicerContext) -> SliceIdList:

        return slice_list_ids(self.db_engine, request)

        return SliceIdList(slice_ids=slice_list_ids(self.db_engine, request))



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

    def ListSlices(self, request : ContextId, context : grpc.ServicerContext) -> SliceList:

        return slice_list_objs(self.db_engine, request)

        return SliceList(slices=slice_list_objs(self.db_engine, request))



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        return slice_get(self.db_engine, request)

        return Slice(**slice_get(self.db_engine, request))



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        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

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

        event_type = EventTypeEnum.EVENTTYPE_UPDATE if updated else EventTypeEnum.EVENTTYPE_CREATE

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

        return SliceId(**slice_id)



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        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

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

        if updated:

            event_type = EventTypeEnum.EVENTTYPE_UPDATE

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

        return SliceId(**slice_id)



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

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

        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})

        slice_id,deleted = slice_delete(self.db_engine, request)

        if deleted:

            event_type = EventTypeEnum.EVENTTYPE_REMOVE

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

        return Empty()



    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

src/context/service/database/ConfigRule.py

+1 −1
Original line number Diff line number Diff line
@@ -51,7 +51,7 @@ def compose_config_rules_data( 
def upsert_config_rules(

    session : Session, config_rules : List[Dict],

    device_uuid : Optional[str] = None, service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None,

) -> bool:

) -> List[bool]:

    # TODO: do not delete all rules; just add-remove as needed

    stmt = delete(ConfigRuleModel)

    if device_uuid  is not None: stmt = stmt.where(ConfigRuleModel.device_uuid  == device_uuid )

src/context/service/database/Connection.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.



import re

import datetime, logging, re

from sqlalchemy.dialects.postgresql import insert

from sqlalchemy.engine import Engine

from sqlalchemy.exc import IntegrityError

src/context/service/database/Constraint.py

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

# limitations under the License.



import datetime, logging

from sqlalchemy import delete

from sqlalchemy.dialects.postgresql import insert

from sqlalchemy.orm import Session
@@ -21,8 +22,10 @@ from common.tools.grpc.Tools import grpc_message_to_json_string 
from .models.ConstraintModel import ConstraintKindEnum, ConstraintModel

from .uuids._Builder import get_uuid_random



LOGGER = logging.getLogger(__name__)



def compose_constraints_data(

    constraints : List[Constraint],

    constraints : List[Constraint], now : datetime.datetime,

    service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None

) -> List[Dict]:

    dict_constraints : List[Dict] = list()
@@ -33,6 +36,8 @@ def compose_constraints_data( 
            'position'       : position,

            'kind'           : ConstraintKindEnum._member_map_.get(str_kind.upper()), # pylint: disable=no-member

            'data'           : grpc_message_to_json_string(getattr(constraint, str_kind, {})),

            'created_at'     : now,

            'updated_at'     : now,

        }

        if service_uuid is not None: dict_constraint['service_uuid'] = service_uuid

        if slice_uuid   is not None: dict_constraint['slice_uuid'  ] = slice_uuid
@@ -42,13 +47,27 @@ def compose_constraints_data( 
def upsert_constraints(

    session : Session, constraints : List[Dict],

    service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None

) -> None:

) -> List[bool]:

    # TODO: do not delete all constraints; just add-remove as needed

    stmt = delete(ConstraintModel)

    if service_uuid is not None: stmt = stmt.where(ConstraintModel.service_uuid == service_uuid)

    if slice_uuid   is not None: stmt = stmt.where(ConstraintModel.slice_uuid   == slice_uuid  )

    session.execute(stmt)



    constraint_updates = []

    if len(constraints) > 0:

        session.execute(insert(ConstraintModel).values(constraints))

        stmt = insert(ConstraintModel).values(constraints)

        #stmt = stmt.on_conflict_do_update(

        #    index_elements=[ConstraintModel.configrule_uuid],

        #    set_=dict(

        #        updated_at = stmt.excluded.updated_at,

        #    )

        #)

        stmt = stmt.returning(ConstraintModel.created_at, ConstraintModel.updated_at)

        constraint_updates = session.execute(stmt).fetchall()



    return constraint_updates





#    def set_constraint(self, db_constraints: ConstraintsModel, grpc_constraint: Constraint, position: int

#    ) -> Tuple[Union_ConstraintModel, bool]:

src/context/service/database/Service.py

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

# limitations under the License.



import datetime, logging

from sqlalchemy.dialects.postgresql import insert

from sqlalchemy.engine import Engine

from sqlalchemy.orm import Session, sessionmaker

from sqlalchemy_cockroachdb import run_transaction

from typing import Dict, List, Optional, Tuple

from common.proto.context_pb2 import ContextId, Service, ServiceId, ServiceIdList, ServiceList

from common.proto.context_pb2 import ContextId, Service, ServiceId

from common.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException

from common.tools.object_factory.Context import json_context_id

from common.tools.object_factory.Service import json_service_id
@@ -30,23 +31,23 @@ from .uuids.Context import context_get_uuid 
from .uuids.EndPoint import endpoint_get_uuid

from .uuids.Service import service_get_uuid



def service_list_ids(db_engine : Engine, request : ContextId) -> ServiceIdList:

LOGGER = logging.getLogger(__name__)



def service_list_ids(db_engine : Engine, request : ContextId) -> List[Dict]:

    context_uuid = context_get_uuid(request, allow_random=False)

    def callback(session : Session) -> List[Dict]:

        obj_list : List[ServiceModel] = session.query(ServiceModel).filter_by(context_uuid=context_uuid).all()

        #.options(selectinload(ContextModel.service)).filter_by(context_uuid=context_uuid).one_or_none()

        return [obj.dump_id() for obj in obj_list]

    return ServiceIdList(service_ids=run_transaction(sessionmaker(bind=db_engine), callback))

    return run_transaction(sessionmaker(bind=db_engine), callback)



def service_list_objs(db_engine : Engine, request : ContextId) -> ServiceList:

def service_list_objs(db_engine : Engine, request : ContextId) -> List[Dict]:

    context_uuid = context_get_uuid(request, allow_random=False)

    def callback(session : Session) -> List[Dict]:

        obj_list : List[ServiceModel] = session.query(ServiceModel).filter_by(context_uuid=context_uuid).all()

        #.options(selectinload(ContextModel.service)).filter_by(context_uuid=context_uuid).one_or_none()

        return [obj.dump() for obj in obj_list]

    return ServiceList(services=run_transaction(sessionmaker(bind=db_engine), callback))

    return run_transaction(sessionmaker(bind=db_engine), callback)



def service_get(db_engine : Engine, request : ServiceId) -> Service:

def service_get(db_engine : Engine, request : ServiceId) -> Dict:

    _,service_uuid = service_get_uuid(request, allow_random=False)

    def callback(session : Session) -> Optional[Dict]:

        obj : Optional[ServiceModel] = session.query(ServiceModel).filter_by(service_uuid=service_uuid).one_or_none()
@@ -59,9 +60,9 @@ def service_get(db_engine : Engine, request : ServiceId) -> Service: 
            'context_uuid generated was: {:s}'.format(context_uuid),

            'service_uuid generated was: {:s}'.format(service_uuid),

        ])

    return Service(**obj)

    return obj



def service_set(db_engine : Engine, request : Service) -> Tuple[ServiceId, bool]:

def service_set(db_engine : Engine, request : Service) -> Tuple[Dict, bool]:

    raw_context_uuid = request.service_id.context_id.context_uuid.uuid

    raw_service_uuid = request.service_id.service_uuid.uuid

    raw_service_name = request.name
@@ -71,6 +72,8 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[ServiceId, bool] 
    service_type = grpc_to_enum__service_type(request.service_type)

    service_status = grpc_to_enum__service_status(request.service_status.service_status)



    now = datetime.datetime.utcnow()



    service_endpoints_data : List[Dict] = list()

    for i,endpoint_id in enumerate(request.service_endpoint_ids):

        endpoint_context_uuid = endpoint_id.topology_id.context_id.context_uuid.uuid
@@ -87,8 +90,8 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[ServiceId, bool] 
            'endpoint_uuid': endpoint_uuid,

        })



    constraints = compose_constraints_data(request.service_constraints, service_uuid=service_uuid)

    config_rules = compose_config_rules_data(request.service_config.config_rules, service_uuid=service_uuid)

    constraints = compose_constraints_data(request.service_constraints, now, service_uuid=service_uuid)

    config_rules = compose_config_rules_data(request.service_config.config_rules, now, service_uuid=service_uuid)



    service_data = [{

        'context_uuid'  : context_uuid,
@@ -96,9 +99,11 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[ServiceId, bool] 
        'service_name'  : service_name,

        'service_type'  : service_type,

        'service_status': service_status,

        'created_at'    : now,

        'updated_at'    : now,

    }]



    def callback(session : Session) -> None:

    def callback(session : Session) -> bool:

        stmt = insert(ServiceModel).values(service_data)

        stmt = stmt.on_conflict_do_update(

            index_elements=[ServiceModel.service_uuid],
@@ -108,7 +113,9 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[ServiceId, bool] 
                service_status = stmt.excluded.service_status,

            )

        )

        session.execute(stmt)

        stmt = stmt.returning(ServiceModel.created_at, ServiceModel.updated_at)

        created_at,updated_at = session.execute(stmt).fetchone()

        updated = updated_at > created_at



        stmt = insert(ServiceEndPointModel).values(service_endpoints_data)

        stmt = stmt.on_conflict_do_nothing(
@@ -116,17 +123,21 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[ServiceId, bool] 
        )

        session.execute(stmt)



        upsert_constraints(session, constraints, service_uuid=service_uuid)

        upsert_config_rules(session, config_rules, service_uuid=service_uuid)

        constraint_updates = upsert_constraints(session, constraints, service_uuid=service_uuid)

        updated = updated or any([(updated_at > created_at) for created_at,updated_at in constraint_updates])



    run_transaction(sessionmaker(bind=db_engine), callback)

    updated = False # TODO: improve and check if created/updated

    return ServiceId(**json_service_id(service_uuid, json_context_id(context_uuid))),updated

        configrule_updates = upsert_config_rules(session, config_rules, service_uuid=service_uuid)

        updated = updated or any([(updated_at > created_at) for created_at,updated_at in configrule_updates])



        return updated



def service_delete(db_engine : Engine, request : ServiceId) -> bool:

    _,service_uuid = service_get_uuid(request, allow_random=False)

    updated = run_transaction(sessionmaker(bind=db_engine), callback)

    return json_service_id(service_uuid, json_context_id(context_uuid)),updated



def service_delete(db_engine : Engine, request : ServiceId) -> Tuple[Dict, bool]:

    context_uuid,service_uuid = service_get_uuid(request, allow_random=False)

    def callback(session : Session) -> bool:

        num_deleted = session.query(ServiceModel).filter_by(service_uuid=service_uuid).delete()

        return num_deleted > 0

    return run_transaction(sessionmaker(bind=db_engine), callback)
 
    deleted = run_transaction(sessionmaker(bind=db_engine), callback)

    return json_service_id(service_uuid, json_context_id(context_uuid)),deleted