Common - Tools: (4f633186) · Commits · TFS / controller

src/common/tools/grpc/Constraints.py

+45 −2

Original line number	Diff line number	Diff line
		@@ -16,9 +16,8 @@
		# Ref: https://datatracker.ietf.org/doc/html/rfc8466


		import json
		from typing import Any, Dict, List, Optional, Tuple
		from common.proto.context_pb2 import Constraint, ConstraintActionEnum, EndPointId
		from common.proto.context_pb2 import Constraint, ConstraintActionEnum, EndPointId, QoSProfileId
		from common.tools.grpc.Tools import grpc_message_to_json_string

		def update_constraint_custom_scalar(
		@@ -81,6 +80,23 @@ def update_constraint_custom_dict(
		constraint.custom.constraint_value = json.dumps(json_constraint_value, sort_keys=True)
		return constraint

		def update_constraint_schedule(
		constraints, start_timestamp : float, duration_days : float,
		new_action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Constraint:
		for constraint in constraints:
		if constraint.WhichOneof('constraint') != 'schedule': continue
		break # found, end loop
		else:
		# not found, add it
		constraint = constraints.add() # pylint: disable=no-member

		constraint.action = new_action

		constraint.schedule.start_timestamp = start_timestamp
		constraint.schedule.duration_days = duration_days
		return constraint

		def update_constraint_endpoint_location(
		constraints, endpoint_id : EndPointId,
		region : Optional[str] = None, gps_position : Optional[Tuple[float, float]] = None,
		@@ -221,6 +237,23 @@ def update_constraint_sla_isolation(
		constraint.sla_isolation.isolation_level.append(isolation_level)
		return constraint

		def update_constraint_qos_profile(
		constraints, qos_profile_id : QoSProfileId, qos_profile_name : str,
		new_action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Constraint:
		for constraint in constraints:
		if constraint.WhichOneof('constraint') != 'qos_profile': continue
		break # found, end loop
		else:
		# not found, add it
		constraint = constraints.add() # pylint: disable=no-member

		constraint.action = new_action

		constraint.qos_profile.qos_profile_id.qos_profile_id.uuid = qos_profile_id.qos_profile_id.uuid
		constraint.qos_profile.qos_profile_name = qos_profile_name
		return constraint

		def copy_constraints(source_constraints, target_constraints):
		for source_constraint in source_constraints:
		constraint_kind = source_constraint.WhichOneof('constraint')
		@@ -240,6 +273,11 @@ def copy_constraints(source_constraints, target_constraints):
		update_constraint_custom_scalar(
		target_constraints, constraint_type, constraint_value, raise_if_differs=raise_if_differs)

		elif constraint_kind == 'schedule':
		start_timestamp = source_constraint.schedule.start_timestamp
		duration_days = source_constraint.schedule.duration_days
		update_constraint_schedule(target_constraints, start_timestamp, duration_days)

		elif constraint_kind == 'endpoint_location':
		endpoint_id = source_constraint.endpoint_location.endpoint_id
		location = source_constraint.endpoint_location.location
		@@ -282,5 +320,10 @@ def copy_constraints(source_constraints, target_constraints):
		isolation_levels = sla_isolation.isolation_level
		update_constraint_sla_isolation(target_constraints, isolation_levels)

		elif constraint_kind == 'qos_profile':
		qos_profile_id = source_constraint.qos_profile.qos_profile_id
		qos_profile_name = source_constraint.qos_profile.qos_profile_name
		update_constraint_qos_profile(target_constraints, qos_profile_id, qos_profile_name)

		else:
		raise NotImplementedError('Constraint({:s})'.format(grpc_message_to_json_string(source_constraint)))

src/common/tools/object_factory/ConfigRule.py

+5 −3

Original line number	Diff line number	Diff line
		@@ -16,12 +16,14 @@ import json
		from typing import Any, Dict, Union
		from common.proto.context_pb2 import ConfigActionEnum

		def json_config_rule(action : ConfigActionEnum, resource_key : str, resource_value : Union[str, Dict[str, Any]]):
		def json_config_rule(
		action : ConfigActionEnum, resource_key : str, resource_value : Union[str, Dict[str, Any]]
		) -> Dict:
		if not isinstance(resource_value, str): resource_value = json.dumps(resource_value, sort_keys=True)
		return {'action': action, 'custom': {'resource_key': resource_key, 'resource_value': resource_value}}

		def json_config_rule_set(resource_key : str, resource_value : Union[str, Dict[str, Any]]):
		def json_config_rule_set(resource_key : str, resource_value : Union[str, Dict[str, Any]]) -> Dict:
		return json_config_rule(ConfigActionEnum.CONFIGACTION_SET, resource_key, resource_value)

		def json_config_rule_delete(resource_key : str, resource_value : Union[str, Dict[str, Any]]):
		def json_config_rule_delete(resource_key : str, resource_value : Union[str, Dict[str, Any]]) -> Dict:
		return json_config_rule(ConfigActionEnum.CONFIGACTION_DELETE, resource_key, resource_value)

src/common/tools/object_factory/Constraint.py

+76 −24

Original line number	Diff line number	Diff line
		@@ -15,44 +15,96 @@
		import json
		from typing import Any, Dict, List, Union

		def json_constraint_custom(constraint_type : str, constraint_value : Union[str, Dict[str, Any]]) -> Dict:
		from common.proto.context_pb2 import ConstraintActionEnum


		def json_constraint_custom(
		constraint_type : str, constraint_value : Union[str, Dict[str, Any]],
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		if not isinstance(constraint_value, str): constraint_value = json.dumps(constraint_value, sort_keys=True)
		return {'custom': {'constraint_type': constraint_type, 'constraint_value': constraint_value}}
		return {'action': action, 'custom': {
		'constraint_type': constraint_type, 'constraint_value': constraint_value
		}}

		def json_constraint_schedule(start_timestamp : float, duration_days : float) -> Dict:
		return {'schedule': {'start_timestamp': start_timestamp, 'duration_days': duration_days}}
		def json_constraint_schedule(
		start_timestamp : float, duration_days : float,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'schedule': {
		'start_timestamp': start_timestamp, 'duration_days': duration_days
		}}

		def json_constraint_endpoint_location_region(endpoint_id : Dict, region : str) -> Dict:
		return {'endpoint_location': {'endpoint_id': endpoint_id, 'location': {'region': region}}}
		def json_constraint_endpoint_location_region(
		endpoint_id : Dict, region : str,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'endpoint_location': {
		'endpoint_id': endpoint_id, 'location': {'region': region}
		}}

		def json_constraint_endpoint_location_gps(endpoint_id : Dict, latitude : float, longitude : float) -> Dict:
		def json_constraint_endpoint_location_gps(
		endpoint_id : Dict, latitude : float, longitude : float,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		gps_position = {'latitude': latitude, 'longitude': longitude}
		return {'endpoint_location': {'endpoint_id': endpoint_id, 'location': {'gps_position': gps_position}}}
		return {'action': action, 'endpoint_location': {
		'endpoint_id': endpoint_id, 'location': {'gps_position': gps_position}
		}}

		def json_constraint_endpoint_priority(endpoint_id : Dict, priority : int) -> Dict:
		return {'endpoint_priority': {'endpoint_id': endpoint_id, 'priority': priority}}
		def json_constraint_endpoint_priority(
		endpoint_id : Dict, priority : int,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'endpoint_priority': {
		'endpoint_id': endpoint_id, 'priority': priority
		}}

		def json_constraint_sla_capacity(capacity_gbps : float) -> Dict:
		return {'sla_capacity': {'capacity_gbps': capacity_gbps}}
		def json_constraint_sla_capacity(
		capacity_gbps : float,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'sla_capacity': {
		'capacity_gbps': capacity_gbps
		}}

		def json_constraint_sla_latency(e2e_latency_ms : float) -> Dict:
		return {'sla_latency': {'e2e_latency_ms': e2e_latency_ms}}
		def json_constraint_sla_latency(
		e2e_latency_ms : float,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'sla_latency': {
		'e2e_latency_ms': e2e_latency_ms
		}}

		def json_constraint_sla_availability(num_disjoint_paths : int, all_active : bool, availability : float) -> Dict:
		return {'sla_availability': {
		def json_constraint_sla_availability(
		num_disjoint_paths : int, all_active : bool, availability : float,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'sla_availability': {
		'num_disjoint_paths': num_disjoint_paths, 'all_active': all_active, 'availability': availability
		}}

		def json_constraint_sla_isolation(isolation_levels : List[int]) -> Dict:
		return {'sla_isolation': {'isolation_level': isolation_levels}}
		def json_constraint_sla_isolation(
		isolation_levels : List[int],
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'sla_isolation': {
		'isolation_level': isolation_levels
		}}

		def json_constraint_exclusions(
		is_permanent : bool = False, device_ids : List[Dict] = [], endpoint_ids : List[Dict] = [],
		link_ids : List[Dict] = []
		link_ids : List[Dict] = [],
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'action': action, 'exclusions': {
		'is_permanent': is_permanent, 'device_ids': device_ids, 'endpoint_ids': endpoint_ids, 'link_ids': link_ids
		}}

		def json_constraint_qos_profile(
		qos_profile_id : Dict, qos_profile_name : int,
		action : ConstraintActionEnum = ConstraintActionEnum.CONSTRAINTACTION_SET
		) -> Dict:
		return {'exclusions': {
		'is_permanent' : is_permanent,
		'device_ids' : device_ids,
		'endpoint_ids' : endpoint_ids,
		'link_ids' : link_ids,
		return {'action': action, 'qos_profile': {
		'qos_profile_id': qos_profile_id, 'qos_profile_name': qos_profile_name
		}}