# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/) # # 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 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, Set, Tuple from common.proto.context_pb2 import ContextId, Slice, SliceId from common.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException from common.tools.object_factory.Context import json_context_id from common.tools.object_factory.Slice import json_slice_id from context.service.database.ConfigRule import compose_config_rules_data, upsert_config_rules from context.service.database.Constraint import compose_constraints_data, upsert_constraints from .models.enums.SliceStatus import grpc_to_enum__slice_status from .models.SliceModel import SliceModel, SliceEndPointModel, SliceServiceModel, SliceSubSliceModel from .uuids.Context import context_get_uuid from .uuids.EndPoint import endpoint_get_uuid from .uuids.Service import service_get_uuid from .uuids.Slice import slice_get_uuid LOGGER = logging.getLogger(__name__) def slice_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[SliceModel] = session.query(SliceModel).filter_by(context_uuid=context_uuid).all() return [obj.dump_id() for obj in obj_list] return run_transaction(sessionmaker(bind=db_engine), callback) def slice_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[SliceModel] = session.query(SliceModel).filter_by(context_uuid=context_uuid).all() return [obj.dump() for obj in obj_list] return run_transaction(sessionmaker(bind=db_engine), callback) def slice_get(db_engine : Engine, request : SliceId) -> Dict: _,slice_uuid = slice_get_uuid(request, allow_random=False) def callback(session : Session) -> Optional[Dict]: obj : Optional[SliceModel] = session.query(SliceModel).filter_by(slice_uuid=slice_uuid).one_or_none() return None if obj is None else obj.dump() obj = run_transaction(sessionmaker(bind=db_engine), callback) if obj is None: context_uuid = context_get_uuid(request.context_id, allow_random=False) raw_slice_uuid = '{:s}/{:s}'.format(request.context_id.context_uuid.uuid, request.slice_uuid.uuid) raise NotFoundException('Slice', raw_slice_uuid, extra_details=[ 'context_uuid generated was: {:s}'.format(context_uuid), 'slice_uuid generated was: {:s}'.format(slice_uuid), ]) return obj def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: raw_context_uuid = request.slice_id.context_id.context_uuid.uuid raw_slice_uuid = request.slice_id.slice_uuid.uuid raw_slice_name = request.name slice_name = raw_slice_uuid if len(raw_slice_name) == 0 else raw_slice_name context_uuid,slice_uuid = slice_get_uuid(request.slice_id, slice_name=slice_name, allow_random=True) slice_status = grpc_to_enum__slice_status(request.slice_status.slice_status) now = datetime.datetime.utcnow() slice_endpoints_data : List[Dict] = list() for i,endpoint_id in enumerate(request.slice_endpoint_ids): endpoint_context_uuid = endpoint_id.topology_id.context_id.context_uuid.uuid if len(endpoint_context_uuid) == 0: endpoint_context_uuid = context_uuid if endpoint_context_uuid not in {raw_context_uuid, context_uuid}: raise InvalidArgumentException( 'request.slice_endpoint_ids[{:d}].topology_id.context_id.context_uuid.uuid'.format(i), endpoint_context_uuid, ['should be == request.slice_id.context_id.context_uuid.uuid({:s})'.format(raw_context_uuid)]) _, _, endpoint_uuid = endpoint_get_uuid(endpoint_id, allow_random=False) slice_endpoints_data.append({ 'slice_uuid' : slice_uuid, 'endpoint_uuid': endpoint_uuid, }) slice_services_data : List[Dict] = list() for i,service_id in enumerate(request.slice_service_ids): _, service_uuid = service_get_uuid(service_id, allow_random=False) slice_services_data.append({ 'slice_uuid' : slice_uuid, 'service_uuid': service_uuid, }) slice_subslices_data : List[Dict] = list() for i,subslice_id in enumerate(request.slice_subslice_ids): _, subslice_uuid = slice_get_uuid(subslice_id, allow_random=False) slice_subslices_data.append({ 'slice_uuid' : slice_uuid, 'subslice_uuid': subslice_uuid, }) constraints = compose_constraints_data(request.slice_constraints, now, slice_uuid=slice_uuid) config_rules = compose_config_rules_data(request.slice_config.config_rules, now, slice_uuid=slice_uuid) slice_data = [{ 'context_uuid' : context_uuid, 'slice_uuid' : slice_uuid, 'slice_name' : slice_name, 'slice_status' : slice_status, 'slice_owner_uuid' : request.slice_owner.owner_uuid.uuid, 'slice_owner_string': request.slice_owner.owner_string, 'created_at' : now, 'updated_at' : now, }] def callback(session : Session) -> bool: stmt = insert(SliceModel).values(slice_data) stmt = stmt.on_conflict_do_update( index_elements=[SliceModel.slice_uuid], set_=dict( slice_name = stmt.excluded.slice_name, slice_status = stmt.excluded.slice_status, updated_at = stmt.excluded.updated_at, slice_owner_uuid = stmt.excluded.slice_owner_uuid, slice_owner_string = stmt.excluded.slice_owner_string, ) ) stmt = stmt.returning(SliceModel.created_at, SliceModel.updated_at) created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at if len(slice_endpoints_data) > 0: stmt = insert(SliceEndPointModel).values(slice_endpoints_data) stmt = stmt.on_conflict_do_nothing( index_elements=[SliceEndPointModel.slice_uuid, SliceEndPointModel.endpoint_uuid] ) session.execute(stmt) if len(slice_services_data) > 0: stmt = insert(SliceServiceModel).values(slice_services_data) stmt = stmt.on_conflict_do_nothing( index_elements=[SliceServiceModel.slice_uuid, SliceServiceModel.service_uuid] ) session.execute(stmt) if len(slice_subslices_data) > 0: stmt = insert(SliceSubSliceModel).values(slice_subslices_data) stmt = stmt.on_conflict_do_nothing( index_elements=[SliceSubSliceModel.slice_uuid, SliceSubSliceModel.subslice_uuid] ) session.execute(stmt) constraint_updates = upsert_constraints(session, constraints, slice_uuid=slice_uuid) updated = updated or any([(updated_at > created_at) for created_at,updated_at in constraint_updates]) configrule_updates = upsert_config_rules(session, config_rules, slice_uuid=slice_uuid) updated = updated or any([(updated_at > created_at) for created_at,updated_at in configrule_updates]) return updated updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_slice_id(slice_uuid, json_context_id(context_uuid)),updated def slice_unset(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: raw_context_uuid = request.slice_id.context_id.context_uuid.uuid raw_slice_uuid = request.slice_id.slice_uuid.uuid raw_slice_name = request.name slice_name = raw_slice_uuid if len(raw_slice_name) == 0 else raw_slice_name context_uuid,slice_uuid = slice_get_uuid(request.slice_id, slice_name=slice_name, allow_random=False) if len(request.slice_constraints) > 0: raise NotImplementedError('UnsetSlice: removal of constraints') if len(request.slice_config.config_rules) > 0: raise NotImplementedError('UnsetSlice: removal of config rules') if len(request.slice_endpoint_ids) > 0: raise NotImplementedError('UnsetSlice: removal of endpoints') slice_endpoint_uuids : Set[str] = set() for i,endpoint_id in enumerate(request.slice_endpoint_ids): endpoint_context_uuid = endpoint_id.topology_id.context_id.context_uuid.uuid if len(endpoint_context_uuid) == 0: endpoint_context_uuid = context_uuid if endpoint_context_uuid not in {raw_context_uuid, context_uuid}: raise InvalidArgumentException( 'request.slice_endpoint_ids[{:d}].topology_id.context_id.context_uuid.uuid'.format(i), endpoint_context_uuid, ['should be == request.slice_id.context_id.context_uuid.uuid({:s})'.format(raw_context_uuid)]) slice_endpoint_uuids.add(endpoint_get_uuid(endpoint_id, allow_random=False)[2]) slice_service_uuids : Set[str] = { service_get_uuid(service_id, allow_random=False)[1] for service_id in request.slice_service_ids } slice_subslice_uuids : Set[str] = { slice_get_uuid(subslice_id, allow_random=False)[1] for subslice_id in request.slice_subslice_ids } def callback(session : Session) -> bool: num_deletes = 0 num_deletes += session.query(SliceServiceModel)\ .filter_by( SliceServiceModel.slice_uuid == slice_uuid, SliceServiceModel.service_uuid.in_(slice_service_uuids) ).delete() num_deletes += session.query(SliceSubSliceModel)\ .filter_by( SliceSubSliceModel.slice_uuid == slice_uuid, SliceSubSliceModel.subslice_uuid.in_(slice_subslice_uuids) ).delete() num_deletes += session.query(SliceEndPointModel)\ .filter_by( SliceEndPointModel.slice_uuid == slice_uuid, SliceEndPointModel.endpoint_uuid.in_(slice_endpoint_uuids) ).delete() return num_deletes > 0 updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_slice_id(slice_uuid, json_context_id(context_uuid)),updated def slice_delete(db_engine : Engine, request : SliceId) -> Tuple[Dict, bool]: context_uuid,slice_uuid = slice_get_uuid(request, allow_random=False) def callback(session : Session) -> bool: num_deleted = session.query(SliceModel).filter_by(slice_uuid=slice_uuid).delete() return num_deleted > 0 deleted = run_transaction(sessionmaker(bind=db_engine), callback) return json_slice_id(slice_uuid, json_context_id(context_uuid)),deleted