Loading src/context/service/database/ConfigRule.py +95 −161 Original line number Diff line number Diff line Loading @@ -12,16 +12,17 @@ # See the License for the specific language governing permissions and # limitations under the License. import datetime, logging import datetime, json, logging from sqlalchemy import delete from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from typing import Dict, List, Optional from typing import Dict, List, Optional, Set from common.proto.context_pb2 import ConfigRule from common.tools.grpc.Tools import grpc_message_to_json_string from .models.enums.ConfigAction import grpc_to_enum__config_action from .models.enums.ConfigAction import ORM_ConfigActionEnum, grpc_to_enum__config_action from .models.ConfigRuleModel import ConfigRuleKindEnum, ConfigRuleModel from .uuids._Builder import get_uuid_random from .uuids._Builder import get_uuid_from_string from .uuids.EndPoint import endpoint_get_uuid LOGGER = logging.getLogger(__name__) Loading @@ -31,173 +32,106 @@ def compose_config_rules_data( ) -> List[Dict]: dict_config_rules : List[Dict] = list() for position,config_rule in enumerate(config_rules): configrule_uuid = get_uuid_random() str_kind = config_rule.WhichOneof('config_rule') kind = ConfigRuleKindEnum._member_map_.get(str_kind.upper()) # pylint: disable=no-member dict_config_rule = { 'configrule_uuid': configrule_uuid, 'position' : position, 'kind' : ConfigRuleKindEnum._member_map_.get(str_kind.upper()), # pylint: disable=no-member 'kind' : kind, 'action' : grpc_to_enum__config_action(config_rule.action), 'data' : grpc_message_to_json_string(getattr(config_rule, str_kind, {})), 'created_at': now, 'updated_at': now, } if device_uuid is not None: dict_config_rule['device_uuid' ] = device_uuid if service_uuid is not None: dict_config_rule['service_uuid'] = service_uuid if slice_uuid is not None: dict_config_rule['slice_uuid' ] = slice_uuid parent_uuid = None if device_uuid is not None: dict_config_rule['device_uuid'] = device_uuid parent_uuid = device_uuid elif service_uuid is not None: dict_config_rule['service_uuid'] = service_uuid parent_uuid = service_uuid elif slice_uuid is not None: dict_config_rule['slice_uuid'] = slice_uuid parent_uuid = slice_uuid else: MSG = 'Parent for ConfigRule({:s}) cannot be identified '+\ '(device_uuid={:s}, service_uuid={:s}, slice_uuid={:s})' str_config_rule = grpc_message_to_json_string(config_rule) raise Exception(MSG.format(str_config_rule, str(device_uuid), str(service_uuid), str(slice_uuid))) configrule_name = None if kind == ConfigRuleKindEnum.CUSTOM: configrule_name = config_rule.custom.resource_key elif kind == ConfigRuleKindEnum.ACL: endpoint_uuid = endpoint_get_uuid(config_rule.acl.endpoint_id, allow_random=False) rule_set_name = config_rule.acl.rule_set.name configrule_name = '{:s}/{:s}'.format(endpoint_uuid, rule_set_name) else: MSG = 'Name for ConfigRule({:s}) cannot be inferred '+\ '(device_uuid={:s}, service_uuid={:s}, slice_uuid={:s})' str_config_rule = grpc_message_to_json_string(config_rule) raise Exception(MSG.format(str_config_rule, str(device_uuid), str(service_uuid), str(slice_uuid))) configrule_uuid = get_uuid_from_string(configrule_name, prefix_for_name=parent_uuid) dict_config_rule['configrule_uuid'] = configrule_uuid dict_config_rules.append(dict_config_rule) return dict_config_rules 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, ) -> List[bool]: # TODO: do not delete all rules; just add-remove as needed ) -> bool: uuids_to_delete : Set[str] = set() uuids_to_upsert : Dict[str, int] = dict() rules_to_upsert : List[Dict] = list() for config_rule in config_rules: if config_rule['action'] == ORM_ConfigActionEnum.SET: configrule_uuid = config_rule['configrule_uuid'] position = uuids_to_upsert.get(configrule_uuid) if position is None: # if not added, add it rules_to_upsert.append(config_rule) uuids_to_upsert[config_rule['configrule_uuid']] = len(rules_to_upsert) - 1 else: # if already added, update occurrence rules_to_upsert[position] = config_rule elif config_rule['action'] == ORM_ConfigActionEnum.DELETE: uuids_to_delete.add(config_rule['configrule_uuid']) else: MSG = 'Action for ConfigRule({:s}) is not supported '+\ '(device_uuid={:s}, service_uuid={:s}, slice_uuid={:s})' str_config_rule = json.dumps(config_rule) raise Exception(MSG.format(str_config_rule, str(device_uuid), str(service_uuid), str(slice_uuid))) #LOGGER.warning('uuids_to_delete={:s}'.format(str(uuids_to_delete))) #LOGGER.warning('rules_to_upsert={:s}'.format(str(rules_to_upsert))) delete_affected = False upsert_affected = False if len(uuids_to_delete) > 0: stmt = delete(ConfigRuleModel) if device_uuid is not None: stmt = stmt.where(ConfigRuleModel.device_uuid == device_uuid ) if service_uuid is not None: stmt = stmt.where(ConfigRuleModel.service_uuid == service_uuid) if slice_uuid is not None: stmt = stmt.where(ConfigRuleModel.slice_uuid == slice_uuid ) session.execute(stmt) stmt = stmt.where(ConfigRuleModel.configrule_uuid.in_(uuids_to_delete)) configrule_deletes = session.execute(stmt)#.fetchall() delete_affected = int(configrule_deletes.rowcount) > 0 configrule_updates = [] if len(config_rules) > 0: stmt = insert(ConfigRuleModel).values(config_rules) #stmt = stmt.on_conflict_do_update( # index_elements=[ConfigRuleModel.configrule_uuid], # set_=dict( # updated_at = stmt.excluded.updated_at, # ) #) if len(rules_to_upsert) > 0: stmt = insert(ConfigRuleModel).values(rules_to_upsert) stmt = stmt.on_conflict_do_update( index_elements=[ConfigRuleModel.configrule_uuid], set_=dict( position = stmt.excluded.position, action = stmt.excluded.action, data = stmt.excluded.data, updated_at = stmt.excluded.updated_at, ) ) stmt = stmt.returning(ConfigRuleModel.created_at, ConfigRuleModel.updated_at) configrule_updates = session.execute(stmt).fetchall() upsert_affected = any([(updated_at > created_at) for created_at,updated_at in configrule_updates]) return configrule_updates #Union_SpecificConfigRule = Union[ # ConfigRuleCustomModel, ConfigRuleAclModel #] # #def set_config_rule( # database : Database, db_config : ConfigModel, position : int, resource_key : str, resource_value : str, #): # -> Tuple[ConfigRuleModel, bool]: # # str_rule_key_hash = fast_hasher(resource_key) # str_config_rule_key = key_to_str([db_config.config_uuid, str_rule_key_hash], separator=':') # # data = {'config_fk': db_config, 'position': position, 'action': ORM_ConfigActionEnum.SET, 'key': resource_key, # 'value': resource_value} # to_add = ConfigRuleModel(**data) # # result = database.create_or_update(to_add) # return result #Tuple_ConfigRuleSpecs = Tuple[Type, str, Dict, ConfigRuleKindEnum] # #def parse_config_rule_custom(database : Database, grpc_config_rule) -> Tuple_ConfigRuleSpecs: # config_rule_class = ConfigRuleCustomModel # str_config_rule_id = grpc_config_rule.custom.resource_key # config_rule_data = { # 'key' : grpc_config_rule.custom.resource_key, # 'value': grpc_config_rule.custom.resource_value, # } # return config_rule_class, str_config_rule_id, config_rule_data, ConfigRuleKindEnum.CUSTOM # #def parse_config_rule_acl(database : Database, grpc_config_rule) -> Tuple_ConfigRuleSpecs: # config_rule_class = ConfigRuleAclModel # grpc_endpoint_id = grpc_config_rule.acl.endpoint_id # grpc_rule_set = grpc_config_rule.acl.rule_set # device_uuid = grpc_endpoint_id.device_id.device_uuid.uuid # endpoint_uuid = grpc_endpoint_id.endpoint_uuid.uuid # str_endpoint_key = '/'.join([device_uuid, endpoint_uuid]) # #str_endpoint_key, db_endpoint = get_endpoint(database, grpc_endpoint_id) # str_config_rule_id = ':'.join([str_endpoint_key, grpc_rule_set.name]) # config_rule_data = { # #'endpoint_fk': db_endpoint, # 'endpoint_id': grpc_message_to_json_string(grpc_endpoint_id), # 'acl_data': grpc_message_to_json_string(grpc_rule_set), # } # return config_rule_class, str_config_rule_id, config_rule_data, ConfigRuleKindEnum.ACL # #CONFIGRULE_PARSERS = { # 'custom': parse_config_rule_custom, # 'acl' : parse_config_rule_acl, #} # #Union_ConfigRuleModel = Union[ # ConfigRuleCustomModel, ConfigRuleAclModel, #] # #def set_config_rule( # database : Database, db_config : ConfigModel, grpc_config_rule : ConfigRule, position : int #) -> Tuple[Union_ConfigRuleModel, bool]: # grpc_config_rule_kind = str(grpc_config_rule.WhichOneof('config_rule')) # parser = CONFIGRULE_PARSERS.get(grpc_config_rule_kind) # if parser is None: # raise NotImplementedError('ConfigRule of kind {:s} is not implemented: {:s}'.format( # grpc_config_rule_kind, grpc_message_to_json_string(grpc_config_rule))) # # # create specific ConfigRule # config_rule_class, str_config_rule_id, config_rule_data, config_rule_kind = parser(database, grpc_config_rule) # str_config_rule_key_hash = fast_hasher(':'.join([config_rule_kind.value, str_config_rule_id])) # str_config_rule_key = key_to_str([db_config.pk, str_config_rule_key_hash], separator=':') # result : Tuple[Union_ConfigRuleModel, bool] = update_or_create_object( # database, config_rule_class, str_config_rule_key, config_rule_data) # db_specific_config_rule, updated = result # # # create generic ConfigRule # config_rule_fk_field_name = 'config_rule_{:s}_fk'.format(config_rule_kind.value) # config_rule_data = { # 'config_fk': db_config, 'kind': config_rule_kind, 'position': position, # 'action': ORM_ConfigActionEnum.SET, # config_rule_fk_field_name: db_specific_config_rule # } # result : Tuple[ConfigRuleModel, bool] = update_or_create_object( # database, ConfigRuleModel, str_config_rule_key, config_rule_data) # db_config_rule, updated = result # # return db_config_rule, updated # #def delete_config_rule( # database : Database, db_config : ConfigModel, grpc_config_rule : ConfigRule #) -> None: # grpc_config_rule_kind = str(grpc_config_rule.WhichOneof('config_rule')) # parser = CONFIGRULE_PARSERS.get(grpc_config_rule_kind) # if parser is None: # raise NotImplementedError('ConfigRule of kind {:s} is not implemented: {:s}'.format( # grpc_config_rule_kind, grpc_message_to_json_string(grpc_config_rule))) # # # delete generic config rules; self deletes specific config rule # _, str_config_rule_id, _, config_rule_kind = parser(database, grpc_config_rule) # str_config_rule_key_hash = fast_hasher(':'.join([config_rule_kind.value, str_config_rule_id])) # str_config_rule_key = key_to_str([db_config.pk, str_config_rule_key_hash], separator=':') # db_config_rule : Optional[ConfigRuleModel] = get_object( # database, ConfigRuleModel, str_config_rule_key, raise_if_not_found=False) # if db_config_rule is None: return # db_config_rule.delete() # #def update_config( # database : Database, db_parent_pk : str, config_name : str, grpc_config_rules #) -> List[Tuple[Union[ConfigModel, ConfigRuleModel], bool]]: # # str_config_key = key_to_str([config_name, db_parent_pk], separator=':') # result : Tuple[ConfigModel, bool] = get_or_create_object(database, ConfigModel, str_config_key) # db_config, created = result # # db_objects = [(db_config, created)] # # for position,grpc_config_rule in enumerate(grpc_config_rules): # action = grpc_to_enum__config_action(grpc_config_rule.action) # # if action == ORM_ConfigActionEnum.SET: # result : Tuple[ConfigRuleModel, bool] = set_config_rule( # database, db_config, grpc_config_rule, position) # db_config_rule, updated = result # db_objects.append((db_config_rule, updated)) # elif action == ORM_ConfigActionEnum.DELETE: # delete_config_rule(database, db_config, grpc_config_rule) # else: # msg = 'Unsupported Action({:s}) for ConfigRule({:s})' # str_action = str(ConfigActionEnum.Name(action)) # str_config_rule = grpc_message_to_json_string(grpc_config_rule) # raise AttributeError(msg.format(str_action, str_config_rule)) # # return db_objects return delete_affected or upsert_affected src/context/service/database/Constraint.py +4 −3 Original line number Diff line number Diff line Loading @@ -47,14 +47,14 @@ def compose_constraints_data( def upsert_constraints( session : Session, constraints : List[Dict], service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None ) -> List[bool]: ) -> 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 = [] changed = False if len(constraints) > 0: stmt = insert(ConstraintModel).values(constraints) #stmt = stmt.on_conflict_do_update( Loading @@ -65,8 +65,9 @@ def upsert_constraints( #) stmt = stmt.returning(ConstraintModel.created_at, ConstraintModel.updated_at) constraint_updates = session.execute(stmt).fetchall() changed = any([(updated_at > created_at) for created_at,updated_at in constraint_updates]) return constraint_updates return changed # def set_constraint(self, db_constraints: ConstraintsModel, grpc_constraint: Constraint, position: int Loading src/context/service/database/Device.py +5 −4 Original line number Diff line number Diff line Loading @@ -20,6 +20,7 @@ from sqlalchemy_cockroachdb import run_transaction from typing import Dict, List, Optional, Set, Tuple from common.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException from common.proto.context_pb2 import Device, DeviceId from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Device import json_device_id from .models.DeviceModel import DeviceModel from .models.EndPointModel import EndPointModel Loading Loading @@ -136,6 +137,7 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[Dict, bool]: created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at updated_endpoints = False if len(endpoints_data) > 0: stmt = insert(EndPointModel).values(endpoints_data) stmt = stmt.on_conflict_do_update( Loading @@ -149,17 +151,16 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[Dict, bool]: ) stmt = stmt.returning(EndPointModel.created_at, EndPointModel.updated_at) endpoint_updates = session.execute(stmt).fetchall() updated = updated or any([(updated_at > created_at) for created_at,updated_at in endpoint_updates]) updated_endpoints = any([(updated_at > created_at) for created_at,updated_at in endpoint_updates]) if len(related_topologies) > 0: session.execute(insert(TopologyDeviceModel).values(related_topologies).on_conflict_do_nothing( index_elements=[TopologyDeviceModel.topology_uuid, TopologyDeviceModel.device_uuid] )) configrule_updates = upsert_config_rules(session, config_rules, device_uuid=device_uuid) updated = updated or any([(updated_at > created_at) for created_at,updated_at in configrule_updates]) changed_config_rules = upsert_config_rules(session, config_rules, device_uuid=device_uuid) return updated return updated or updated_endpoints or changed_config_rules updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_device_id(device_uuid),updated Loading src/context/service/database/Service.py +4 −6 Original line number Diff line number Diff line Loading @@ -118,6 +118,7 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[Dict, bool]: created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at # TODO: check if endpoints are changed if len(service_endpoints_data) > 0: stmt = insert(ServiceEndPointModel).values(service_endpoints_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -125,13 +126,10 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[Dict, bool]: ) session.execute(stmt) 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]) changed_constraints = upsert_constraints(session, constraints, service_uuid=service_uuid) changed_config_rules = upsert_config_rules(session, config_rules, service_uuid=service_uuid) 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 return updated or changed_constraints or changed_config_rules updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_service_id(service_uuid, json_context_id(context_uuid)),updated Loading src/context/service/database/Slice.py +6 −6 Original line number Diff line number Diff line Loading @@ -136,6 +136,7 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at # TODO: check if endpoints are changed if len(slice_endpoints_data) > 0: stmt = insert(SliceEndPointModel).values(slice_endpoints_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -143,6 +144,7 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: ) session.execute(stmt) # TODO: check if services are changed if len(slice_services_data) > 0: stmt = insert(SliceServiceModel).values(slice_services_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -150,6 +152,7 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: ) session.execute(stmt) # TODO: check if subslices are changed if len(slice_subslices_data) > 0: stmt = insert(SliceSubSliceModel).values(slice_subslices_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -157,13 +160,10 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: ) 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]) changed_constraints = upsert_constraints(session, constraints, slice_uuid=slice_uuid) changed_config_rules = upsert_config_rules(session, config_rules, slice_uuid=slice_uuid) 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 return updated or changed_constraints or changed_config_rules updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_slice_id(slice_uuid, json_context_id(context_uuid)),updated Loading Loading
src/context/service/database/ConfigRule.py +95 −161 Original line number Diff line number Diff line Loading @@ -12,16 +12,17 @@ # See the License for the specific language governing permissions and # limitations under the License. import datetime, logging import datetime, json, logging from sqlalchemy import delete from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session from typing import Dict, List, Optional from typing import Dict, List, Optional, Set from common.proto.context_pb2 import ConfigRule from common.tools.grpc.Tools import grpc_message_to_json_string from .models.enums.ConfigAction import grpc_to_enum__config_action from .models.enums.ConfigAction import ORM_ConfigActionEnum, grpc_to_enum__config_action from .models.ConfigRuleModel import ConfigRuleKindEnum, ConfigRuleModel from .uuids._Builder import get_uuid_random from .uuids._Builder import get_uuid_from_string from .uuids.EndPoint import endpoint_get_uuid LOGGER = logging.getLogger(__name__) Loading @@ -31,173 +32,106 @@ def compose_config_rules_data( ) -> List[Dict]: dict_config_rules : List[Dict] = list() for position,config_rule in enumerate(config_rules): configrule_uuid = get_uuid_random() str_kind = config_rule.WhichOneof('config_rule') kind = ConfigRuleKindEnum._member_map_.get(str_kind.upper()) # pylint: disable=no-member dict_config_rule = { 'configrule_uuid': configrule_uuid, 'position' : position, 'kind' : ConfigRuleKindEnum._member_map_.get(str_kind.upper()), # pylint: disable=no-member 'kind' : kind, 'action' : grpc_to_enum__config_action(config_rule.action), 'data' : grpc_message_to_json_string(getattr(config_rule, str_kind, {})), 'created_at': now, 'updated_at': now, } if device_uuid is not None: dict_config_rule['device_uuid' ] = device_uuid if service_uuid is not None: dict_config_rule['service_uuid'] = service_uuid if slice_uuid is not None: dict_config_rule['slice_uuid' ] = slice_uuid parent_uuid = None if device_uuid is not None: dict_config_rule['device_uuid'] = device_uuid parent_uuid = device_uuid elif service_uuid is not None: dict_config_rule['service_uuid'] = service_uuid parent_uuid = service_uuid elif slice_uuid is not None: dict_config_rule['slice_uuid'] = slice_uuid parent_uuid = slice_uuid else: MSG = 'Parent for ConfigRule({:s}) cannot be identified '+\ '(device_uuid={:s}, service_uuid={:s}, slice_uuid={:s})' str_config_rule = grpc_message_to_json_string(config_rule) raise Exception(MSG.format(str_config_rule, str(device_uuid), str(service_uuid), str(slice_uuid))) configrule_name = None if kind == ConfigRuleKindEnum.CUSTOM: configrule_name = config_rule.custom.resource_key elif kind == ConfigRuleKindEnum.ACL: endpoint_uuid = endpoint_get_uuid(config_rule.acl.endpoint_id, allow_random=False) rule_set_name = config_rule.acl.rule_set.name configrule_name = '{:s}/{:s}'.format(endpoint_uuid, rule_set_name) else: MSG = 'Name for ConfigRule({:s}) cannot be inferred '+\ '(device_uuid={:s}, service_uuid={:s}, slice_uuid={:s})' str_config_rule = grpc_message_to_json_string(config_rule) raise Exception(MSG.format(str_config_rule, str(device_uuid), str(service_uuid), str(slice_uuid))) configrule_uuid = get_uuid_from_string(configrule_name, prefix_for_name=parent_uuid) dict_config_rule['configrule_uuid'] = configrule_uuid dict_config_rules.append(dict_config_rule) return dict_config_rules 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, ) -> List[bool]: # TODO: do not delete all rules; just add-remove as needed ) -> bool: uuids_to_delete : Set[str] = set() uuids_to_upsert : Dict[str, int] = dict() rules_to_upsert : List[Dict] = list() for config_rule in config_rules: if config_rule['action'] == ORM_ConfigActionEnum.SET: configrule_uuid = config_rule['configrule_uuid'] position = uuids_to_upsert.get(configrule_uuid) if position is None: # if not added, add it rules_to_upsert.append(config_rule) uuids_to_upsert[config_rule['configrule_uuid']] = len(rules_to_upsert) - 1 else: # if already added, update occurrence rules_to_upsert[position] = config_rule elif config_rule['action'] == ORM_ConfigActionEnum.DELETE: uuids_to_delete.add(config_rule['configrule_uuid']) else: MSG = 'Action for ConfigRule({:s}) is not supported '+\ '(device_uuid={:s}, service_uuid={:s}, slice_uuid={:s})' str_config_rule = json.dumps(config_rule) raise Exception(MSG.format(str_config_rule, str(device_uuid), str(service_uuid), str(slice_uuid))) #LOGGER.warning('uuids_to_delete={:s}'.format(str(uuids_to_delete))) #LOGGER.warning('rules_to_upsert={:s}'.format(str(rules_to_upsert))) delete_affected = False upsert_affected = False if len(uuids_to_delete) > 0: stmt = delete(ConfigRuleModel) if device_uuid is not None: stmt = stmt.where(ConfigRuleModel.device_uuid == device_uuid ) if service_uuid is not None: stmt = stmt.where(ConfigRuleModel.service_uuid == service_uuid) if slice_uuid is not None: stmt = stmt.where(ConfigRuleModel.slice_uuid == slice_uuid ) session.execute(stmt) stmt = stmt.where(ConfigRuleModel.configrule_uuid.in_(uuids_to_delete)) configrule_deletes = session.execute(stmt)#.fetchall() delete_affected = int(configrule_deletes.rowcount) > 0 configrule_updates = [] if len(config_rules) > 0: stmt = insert(ConfigRuleModel).values(config_rules) #stmt = stmt.on_conflict_do_update( # index_elements=[ConfigRuleModel.configrule_uuid], # set_=dict( # updated_at = stmt.excluded.updated_at, # ) #) if len(rules_to_upsert) > 0: stmt = insert(ConfigRuleModel).values(rules_to_upsert) stmt = stmt.on_conflict_do_update( index_elements=[ConfigRuleModel.configrule_uuid], set_=dict( position = stmt.excluded.position, action = stmt.excluded.action, data = stmt.excluded.data, updated_at = stmt.excluded.updated_at, ) ) stmt = stmt.returning(ConfigRuleModel.created_at, ConfigRuleModel.updated_at) configrule_updates = session.execute(stmt).fetchall() upsert_affected = any([(updated_at > created_at) for created_at,updated_at in configrule_updates]) return configrule_updates #Union_SpecificConfigRule = Union[ # ConfigRuleCustomModel, ConfigRuleAclModel #] # #def set_config_rule( # database : Database, db_config : ConfigModel, position : int, resource_key : str, resource_value : str, #): # -> Tuple[ConfigRuleModel, bool]: # # str_rule_key_hash = fast_hasher(resource_key) # str_config_rule_key = key_to_str([db_config.config_uuid, str_rule_key_hash], separator=':') # # data = {'config_fk': db_config, 'position': position, 'action': ORM_ConfigActionEnum.SET, 'key': resource_key, # 'value': resource_value} # to_add = ConfigRuleModel(**data) # # result = database.create_or_update(to_add) # return result #Tuple_ConfigRuleSpecs = Tuple[Type, str, Dict, ConfigRuleKindEnum] # #def parse_config_rule_custom(database : Database, grpc_config_rule) -> Tuple_ConfigRuleSpecs: # config_rule_class = ConfigRuleCustomModel # str_config_rule_id = grpc_config_rule.custom.resource_key # config_rule_data = { # 'key' : grpc_config_rule.custom.resource_key, # 'value': grpc_config_rule.custom.resource_value, # } # return config_rule_class, str_config_rule_id, config_rule_data, ConfigRuleKindEnum.CUSTOM # #def parse_config_rule_acl(database : Database, grpc_config_rule) -> Tuple_ConfigRuleSpecs: # config_rule_class = ConfigRuleAclModel # grpc_endpoint_id = grpc_config_rule.acl.endpoint_id # grpc_rule_set = grpc_config_rule.acl.rule_set # device_uuid = grpc_endpoint_id.device_id.device_uuid.uuid # endpoint_uuid = grpc_endpoint_id.endpoint_uuid.uuid # str_endpoint_key = '/'.join([device_uuid, endpoint_uuid]) # #str_endpoint_key, db_endpoint = get_endpoint(database, grpc_endpoint_id) # str_config_rule_id = ':'.join([str_endpoint_key, grpc_rule_set.name]) # config_rule_data = { # #'endpoint_fk': db_endpoint, # 'endpoint_id': grpc_message_to_json_string(grpc_endpoint_id), # 'acl_data': grpc_message_to_json_string(grpc_rule_set), # } # return config_rule_class, str_config_rule_id, config_rule_data, ConfigRuleKindEnum.ACL # #CONFIGRULE_PARSERS = { # 'custom': parse_config_rule_custom, # 'acl' : parse_config_rule_acl, #} # #Union_ConfigRuleModel = Union[ # ConfigRuleCustomModel, ConfigRuleAclModel, #] # #def set_config_rule( # database : Database, db_config : ConfigModel, grpc_config_rule : ConfigRule, position : int #) -> Tuple[Union_ConfigRuleModel, bool]: # grpc_config_rule_kind = str(grpc_config_rule.WhichOneof('config_rule')) # parser = CONFIGRULE_PARSERS.get(grpc_config_rule_kind) # if parser is None: # raise NotImplementedError('ConfigRule of kind {:s} is not implemented: {:s}'.format( # grpc_config_rule_kind, grpc_message_to_json_string(grpc_config_rule))) # # # create specific ConfigRule # config_rule_class, str_config_rule_id, config_rule_data, config_rule_kind = parser(database, grpc_config_rule) # str_config_rule_key_hash = fast_hasher(':'.join([config_rule_kind.value, str_config_rule_id])) # str_config_rule_key = key_to_str([db_config.pk, str_config_rule_key_hash], separator=':') # result : Tuple[Union_ConfigRuleModel, bool] = update_or_create_object( # database, config_rule_class, str_config_rule_key, config_rule_data) # db_specific_config_rule, updated = result # # # create generic ConfigRule # config_rule_fk_field_name = 'config_rule_{:s}_fk'.format(config_rule_kind.value) # config_rule_data = { # 'config_fk': db_config, 'kind': config_rule_kind, 'position': position, # 'action': ORM_ConfigActionEnum.SET, # config_rule_fk_field_name: db_specific_config_rule # } # result : Tuple[ConfigRuleModel, bool] = update_or_create_object( # database, ConfigRuleModel, str_config_rule_key, config_rule_data) # db_config_rule, updated = result # # return db_config_rule, updated # #def delete_config_rule( # database : Database, db_config : ConfigModel, grpc_config_rule : ConfigRule #) -> None: # grpc_config_rule_kind = str(grpc_config_rule.WhichOneof('config_rule')) # parser = CONFIGRULE_PARSERS.get(grpc_config_rule_kind) # if parser is None: # raise NotImplementedError('ConfigRule of kind {:s} is not implemented: {:s}'.format( # grpc_config_rule_kind, grpc_message_to_json_string(grpc_config_rule))) # # # delete generic config rules; self deletes specific config rule # _, str_config_rule_id, _, config_rule_kind = parser(database, grpc_config_rule) # str_config_rule_key_hash = fast_hasher(':'.join([config_rule_kind.value, str_config_rule_id])) # str_config_rule_key = key_to_str([db_config.pk, str_config_rule_key_hash], separator=':') # db_config_rule : Optional[ConfigRuleModel] = get_object( # database, ConfigRuleModel, str_config_rule_key, raise_if_not_found=False) # if db_config_rule is None: return # db_config_rule.delete() # #def update_config( # database : Database, db_parent_pk : str, config_name : str, grpc_config_rules #) -> List[Tuple[Union[ConfigModel, ConfigRuleModel], bool]]: # # str_config_key = key_to_str([config_name, db_parent_pk], separator=':') # result : Tuple[ConfigModel, bool] = get_or_create_object(database, ConfigModel, str_config_key) # db_config, created = result # # db_objects = [(db_config, created)] # # for position,grpc_config_rule in enumerate(grpc_config_rules): # action = grpc_to_enum__config_action(grpc_config_rule.action) # # if action == ORM_ConfigActionEnum.SET: # result : Tuple[ConfigRuleModel, bool] = set_config_rule( # database, db_config, grpc_config_rule, position) # db_config_rule, updated = result # db_objects.append((db_config_rule, updated)) # elif action == ORM_ConfigActionEnum.DELETE: # delete_config_rule(database, db_config, grpc_config_rule) # else: # msg = 'Unsupported Action({:s}) for ConfigRule({:s})' # str_action = str(ConfigActionEnum.Name(action)) # str_config_rule = grpc_message_to_json_string(grpc_config_rule) # raise AttributeError(msg.format(str_action, str_config_rule)) # # return db_objects return delete_affected or upsert_affected
src/context/service/database/Constraint.py +4 −3 Original line number Diff line number Diff line Loading @@ -47,14 +47,14 @@ def compose_constraints_data( def upsert_constraints( session : Session, constraints : List[Dict], service_uuid : Optional[str] = None, slice_uuid : Optional[str] = None ) -> List[bool]: ) -> 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 = [] changed = False if len(constraints) > 0: stmt = insert(ConstraintModel).values(constraints) #stmt = stmt.on_conflict_do_update( Loading @@ -65,8 +65,9 @@ def upsert_constraints( #) stmt = stmt.returning(ConstraintModel.created_at, ConstraintModel.updated_at) constraint_updates = session.execute(stmt).fetchall() changed = any([(updated_at > created_at) for created_at,updated_at in constraint_updates]) return constraint_updates return changed # def set_constraint(self, db_constraints: ConstraintsModel, grpc_constraint: Constraint, position: int Loading
src/context/service/database/Device.py +5 −4 Original line number Diff line number Diff line Loading @@ -20,6 +20,7 @@ from sqlalchemy_cockroachdb import run_transaction from typing import Dict, List, Optional, Set, Tuple from common.method_wrappers.ServiceExceptions import InvalidArgumentException, NotFoundException from common.proto.context_pb2 import Device, DeviceId from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.Device import json_device_id from .models.DeviceModel import DeviceModel from .models.EndPointModel import EndPointModel Loading Loading @@ -136,6 +137,7 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[Dict, bool]: created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at updated_endpoints = False if len(endpoints_data) > 0: stmt = insert(EndPointModel).values(endpoints_data) stmt = stmt.on_conflict_do_update( Loading @@ -149,17 +151,16 @@ def device_set(db_engine : Engine, request : Device) -> Tuple[Dict, bool]: ) stmt = stmt.returning(EndPointModel.created_at, EndPointModel.updated_at) endpoint_updates = session.execute(stmt).fetchall() updated = updated or any([(updated_at > created_at) for created_at,updated_at in endpoint_updates]) updated_endpoints = any([(updated_at > created_at) for created_at,updated_at in endpoint_updates]) if len(related_topologies) > 0: session.execute(insert(TopologyDeviceModel).values(related_topologies).on_conflict_do_nothing( index_elements=[TopologyDeviceModel.topology_uuid, TopologyDeviceModel.device_uuid] )) configrule_updates = upsert_config_rules(session, config_rules, device_uuid=device_uuid) updated = updated or any([(updated_at > created_at) for created_at,updated_at in configrule_updates]) changed_config_rules = upsert_config_rules(session, config_rules, device_uuid=device_uuid) return updated return updated or updated_endpoints or changed_config_rules updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_device_id(device_uuid),updated Loading
src/context/service/database/Service.py +4 −6 Original line number Diff line number Diff line Loading @@ -118,6 +118,7 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[Dict, bool]: created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at # TODO: check if endpoints are changed if len(service_endpoints_data) > 0: stmt = insert(ServiceEndPointModel).values(service_endpoints_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -125,13 +126,10 @@ def service_set(db_engine : Engine, request : Service) -> Tuple[Dict, bool]: ) session.execute(stmt) 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]) changed_constraints = upsert_constraints(session, constraints, service_uuid=service_uuid) changed_config_rules = upsert_config_rules(session, config_rules, service_uuid=service_uuid) 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 return updated or changed_constraints or changed_config_rules updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_service_id(service_uuid, json_context_id(context_uuid)),updated Loading
src/context/service/database/Slice.py +6 −6 Original line number Diff line number Diff line Loading @@ -136,6 +136,7 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: created_at,updated_at = session.execute(stmt).fetchone() updated = updated_at > created_at # TODO: check if endpoints are changed if len(slice_endpoints_data) > 0: stmt = insert(SliceEndPointModel).values(slice_endpoints_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -143,6 +144,7 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: ) session.execute(stmt) # TODO: check if services are changed if len(slice_services_data) > 0: stmt = insert(SliceServiceModel).values(slice_services_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -150,6 +152,7 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: ) session.execute(stmt) # TODO: check if subslices are changed if len(slice_subslices_data) > 0: stmt = insert(SliceSubSliceModel).values(slice_subslices_data) stmt = stmt.on_conflict_do_nothing( Loading @@ -157,13 +160,10 @@ def slice_set(db_engine : Engine, request : Slice) -> Tuple[Dict, bool]: ) 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]) changed_constraints = upsert_constraints(session, constraints, slice_uuid=slice_uuid) changed_config_rules = upsert_config_rules(session, config_rules, slice_uuid=slice_uuid) 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 return updated or changed_constraints or changed_config_rules updated = run_transaction(sessionmaker(bind=db_engine), callback) return json_slice_id(slice_uuid, json_context_id(context_uuid)),updated Loading
