Loading src/tests/tools/load_gen/Constants.py +5 −3 Original line number Diff line number Diff line Loading @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. SERVICE_TYPE_L2NM = 'l2nm' SERVICE_TYPE_L3NM = 'l3nm' SERVICE_TYPE_TAPI = 'tapi' REQUEST_TYPE_SERVICE_L2NM = 'svc-l2nm' REQUEST_TYPE_SERVICE_L3NM = 'svc-l3nm' REQUEST_TYPE_SERVICE_TAPI = 'svc-tapi' REQUEST_TYPE_SLICE_L2NM = 'slc-l2nm' REQUEST_TYPE_SLICE_L3NM = 'slc-l3nm' src/tests/tools/load_gen/Parameters.py +5 −5 Original line number Diff line number Diff line Loading @@ -16,12 +16,12 @@ from typing import List, Optional class Parameters: def __init__( self, num_services : int, service_types : List[str], offered_load : Optional[float] = None, self, num_requests : int, request_types : List[str], offered_load : Optional[float] = None, inter_arrival_time : Optional[float] = None, holding_time : Optional[float] = None, dry_mode : bool = False ) -> None: self._num_services = num_services self._service_types = service_types self._num_requests = num_requests self._request_types = request_types self._offered_load = offered_load self._inter_arrival_time = inter_arrival_time self._holding_time = holding_time Loading @@ -35,10 +35,10 @@ class Parameters: self._holding_time = self._offered_load * self._inter_arrival_time @property def num_services(self): return self._num_services def num_requests(self): return self._num_requests @property def service_types(self): return self._service_types def request_types(self): return self._request_types @property def offered_load(self): return self._offered_load Loading src/tests/tools/load_gen/ServiceGenerator.py→src/tests/tools/load_gen/RequestGenerator.py +139 −31 Original line number Diff line number Diff line Loading @@ -21,8 +21,11 @@ from common.tools.object_factory.Device import json_device_id from common.tools.object_factory.EndPoint import json_endpoint_id from common.tools.object_factory.Service import ( json_service_l2nm_planned, json_service_l3nm_planned, json_service_tapi_planned) from common.tools.object_factory.Slice import json_slice from context.client.ContextClient import ContextClient from .Constants import SERVICE_TYPE_L2NM, SERVICE_TYPE_L3NM, SERVICE_TYPE_TAPI from .Constants import ( REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI, REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM) from .Parameters import Parameters LOGGER = logging.getLogger(__name__) Loading @@ -32,11 +35,11 @@ ENDPOINT_COMPATIBILITY = { 'PHOTONIC_MEDIA:DWDM:G_50GHZ:INPUT' : 'PHOTONIC_MEDIA:DWDM:G_50GHZ:OUTPUT', } class ServiceGenerator: class RequestGenerator: def __init__(self, parameters : Parameters) -> None: self._parameters = parameters self._lock = threading.Lock() self._num_services = 0 self._num_requests = 0 self._available_device_endpoints : Dict[str, Set[str]] = dict() self._used_device_endpoints : Dict[str, Dict[str, str]] = dict() self._endpoint_ids_to_types : Dict[Tuple[str, str], str] = dict() Loading Loading @@ -79,7 +82,7 @@ class ServiceGenerator: endpoints_for_type.discard(endpoint_key) @property def num_services_generated(self): return self._num_services def num_requests_generated(self): return self._num_requests def dump_state(self) -> None: with self._lock: Loading Loading @@ -139,19 +142,26 @@ class ServiceGenerator: self._used_device_endpoints.setdefault(device_uuid, set()).pop(endpoint_uuid, None) self._available_device_endpoints.setdefault(device_uuid, set()).add(endpoint_uuid) def compose_service(self) -> Optional[Dict]: def compose_request(self) -> Optional[Dict]: with self._lock: self._num_services += 1 num_service = self._num_services #service_uuid = str(uuid.uuid4()) service_uuid = 'svc_{:d}'.format(num_service) self._num_requests += 1 num_request = self._num_requests # choose service type service_type = random.choice(self._parameters.service_types) #request_uuid = str(uuid.uuid4()) request_uuid = 'svc_{:d}'.format(num_request) # choose request type request_type = random.choice(self._parameters.request_types) if request_type in {REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI}: return self._compose_service(num_request, request_uuid, request_type) elif request_type in {REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM}: return self._compose_slice(num_request, request_uuid, request_type) def _compose_service(self, num_request : int, request_uuid : str, request_type : str) -> Optional[Dict]: # choose source endpoint src_endpoint_types = set(ENDPOINT_COMPATIBILITY.keys()) if service_type in {SERVICE_TYPE_TAPI} else None src = self._use_device_endpoint(service_uuid, endpoint_types=src_endpoint_types) src_endpoint_types = set(ENDPOINT_COMPATIBILITY.keys()) if request_type in {REQUEST_TYPE_SERVICE_TAPI} else None src = self._use_device_endpoint(request_uuid, endpoint_types=src_endpoint_types) if src is None: LOGGER.warning('>> No source endpoint is available') return None Loading @@ -160,14 +170,14 @@ class ServiceGenerator: # identify compatible destination endpoint types src_endpoint_type = self._endpoint_ids_to_types.get((src_device_uuid,src_endpoint_uuid)) dst_endpoint_type = ENDPOINT_COMPATIBILITY.get(src_endpoint_type) dst_endpoint_types = {dst_endpoint_type} if service_type in {SERVICE_TYPE_TAPI} else None dst_endpoint_types = {dst_endpoint_type} if request_type in {REQUEST_TYPE_SERVICE_TAPI} else None # identify expluded destination devices exclude_device_uuids = {} if service_type in {SERVICE_TYPE_TAPI} else {src_device_uuid} # identify excluded destination devices exclude_device_uuids = {} if request_type in {REQUEST_TYPE_SERVICE_TAPI} else {src_device_uuid} # choose feasible destination endpoint dst = self._use_device_endpoint( service_uuid, endpoint_types=dst_endpoint_types, exclude_device_uuids=exclude_device_uuids) request_uuid, endpoint_types=dst_endpoint_types, exclude_device_uuids=exclude_device_uuids) # if destination endpoint not found, release source, and terminate current service generation if dst is None: Loading @@ -182,12 +192,12 @@ class ServiceGenerator: json_endpoint_id(json_device_id(dst_device_uuid), dst_endpoint_uuid), ] if service_type == SERVICE_TYPE_L2NM: if request_type == REQUEST_TYPE_SERVICE_L2NM: constraints = [ json_constraint_custom('bandwidth[gbps]', '10.0'), json_constraint_custom('latency[ms]', '20.0'), ] vlan_id = num_service % 1000 vlan_id = num_request % 1000 circuit_id = '{:03d}'.format(vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) dst_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) Loading @@ -211,15 +221,15 @@ class ServiceGenerator: }), ] return json_service_l2nm_planned( service_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) request_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) elif service_type == SERVICE_TYPE_L3NM: elif request_type == REQUEST_TYPE_SERVICE_L3NM: constraints = [ json_constraint_custom('bandwidth[gbps]', '10.0'), json_constraint_custom('latency[ms]', '20.0'), ] vlan_id = num_service % 1000 bgp_as = 60000 + (num_service % 10000) vlan_id = num_request % 1000 bgp_as = 60000 + (num_request % 10000) bgp_route_target = '{:5d}:{:03d}'.format(bgp_as, 333) route_distinguisher = '{:5d}:{:03d}'.format(bgp_as, vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) Loading Loading @@ -250,9 +260,9 @@ class ServiceGenerator: }), ] return json_service_l3nm_planned( service_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) request_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) elif service_type == SERVICE_TYPE_TAPI: elif request_type == REQUEST_TYPE_SERVICE_TAPI: config_rules = [ json_config_rule_set('/settings', { 'capacity_value' : 50.0, Loading @@ -263,10 +273,108 @@ class ServiceGenerator: }), ] return json_service_tapi_planned( service_uuid, endpoint_ids=endpoint_ids, constraints=[], config_rules=config_rules) request_uuid, endpoint_ids=endpoint_ids, constraints=[], config_rules=config_rules) def _compose_slice(self, num_request : int, request_uuid : str, request_type : str) -> Optional[Dict]: # choose source endpoint src = self._use_device_endpoint(request_uuid) if src is None: LOGGER.warning('>> No source endpoint is available') return None src_device_uuid,src_endpoint_uuid = src # identify excluded destination devices exclude_device_uuids = {} if request_type in {REQUEST_TYPE_SERVICE_TAPI} else {src_device_uuid} # choose feasible destination endpoint dst = self._use_device_endpoint(request_uuid, exclude_device_uuids=exclude_device_uuids) def release_service(self, json_service : Dict) -> None: for endpoint_id in json_service['service_endpoint_ids']: # if destination endpoint not found, release source, and terminate current service generation if dst is None: LOGGER.warning('>> No destination endpoint is available') self._release_device_endpoint(src_device_uuid, src_endpoint_uuid) return None # compose endpoints dst_device_uuid,dst_endpoint_uuid = dst endpoint_ids = [ json_endpoint_id(json_device_id(src_device_uuid), src_endpoint_uuid), json_endpoint_id(json_device_id(dst_device_uuid), dst_endpoint_uuid), ] constraints = [ json_constraint_custom('bandwidth[gbps]', '10.0'), json_constraint_custom('latency[ms]', '20.0'), ] if request_type == REQUEST_TYPE_SLICE_L2NM: vlan_id = num_request % 1000 circuit_id = '{:03d}'.format(vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) dst_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) config_rules = [ json_config_rule_set('/settings', { 'mtu': 1512 }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(src_device_uuid, src_endpoint_uuid), { 'router_id': src_router_id, 'sub_interface_index': vlan_id, 'vlan_id': vlan_id, 'remote_router': dst_router_id, 'circuit_id': circuit_id, }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(dst_device_uuid, dst_endpoint_uuid), { 'router_id': dst_router_id, 'sub_interface_index': vlan_id, 'vlan_id': vlan_id, 'remote_router': src_router_id, 'circuit_id': circuit_id, }), ] elif request_type == REQUEST_TYPE_SLICE_L3NM: vlan_id = num_request % 1000 bgp_as = 60000 + (num_request % 10000) bgp_route_target = '{:5d}:{:03d}'.format(bgp_as, 333) route_distinguisher = '{:5d}:{:03d}'.format(bgp_as, vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) dst_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) src_address_ip = '.'.join([src_device_uuid.replace('R', ''), '0'] + src_endpoint_uuid.split('/')) dst_address_ip = '.'.join([dst_device_uuid.replace('R', ''), '0'] + dst_endpoint_uuid.split('/')) config_rules = [ json_config_rule_set('/settings', { 'mtu' : 1512, 'bgp_as' : bgp_as, 'bgp_route_target': bgp_route_target, }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(src_device_uuid, src_endpoint_uuid), { 'router_id' : src_router_id, 'route_distinguisher': route_distinguisher, 'sub_interface_index': vlan_id, 'vlan_id' : vlan_id, 'address_ip' : src_address_ip, 'address_prefix' : 16, }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(dst_device_uuid, dst_endpoint_uuid), { 'router_id' : dst_router_id, 'route_distinguisher': route_distinguisher, 'sub_interface_index': vlan_id, 'vlan_id' : vlan_id, 'address_ip' : dst_address_ip, 'address_prefix' : 16, }), ] return json_slice( request_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) def release_request(self, json_request : Dict) -> None: if 'service_id' in json_request: for endpoint_id in json_request['service_endpoint_ids']: device_uuid = endpoint_id['device_id']['device_uuid']['uuid'] endpoint_uuid = endpoint_id['endpoint_uuid']['uuid'] self._release_device_endpoint(device_uuid, endpoint_uuid) elif 'slice_id' in json_request: for endpoint_id in json_request['slice_endpoint_ids']: device_uuid = endpoint_id['device_id']['device_uuid']['uuid'] endpoint_uuid = endpoint_id['endpoint_uuid']['uuid'] self._release_device_endpoint(device_uuid, endpoint_uuid) src/tests/tools/load_gen/ServiceScheduler.py→src/tests/tools/load_gen/RequestScheduler.py +144 −0 Original line number Diff line number Diff line Loading @@ -18,18 +18,19 @@ from apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.schedulers.blocking import BlockingScheduler from typing import Dict from common.proto.context_pb2 import Service, ServiceId from common.proto.context_pb2 import Service, ServiceId, Slice, SliceId from service.client.ServiceClient import ServiceClient from slice.client.SliceClient import SliceClient from .Parameters import Parameters from .ServiceGenerator import ServiceGenerator from .RequestGenerator import RequestGenerator logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING) logging.getLogger('apscheduler.scheduler').setLevel(logging.WARNING) LOGGER = logging.getLogger(__name__) class ServiceScheduler: def __init__(self, parameters : Parameters, service_generator : ServiceGenerator) -> None: class RequestScheduler: def __init__(self, parameters : Parameters, generator : RequestGenerator) -> None: self._scheduler = BlockingScheduler() self._scheduler.configure( jobstores = {'default': MemoryJobStore()}, Loading @@ -41,66 +42,103 @@ class ServiceScheduler: }, timezone=pytz.utc) self._parameters = parameters self._service_generator = service_generator self._generator = generator def _schedule_service_setup(self) -> None: if self._service_generator.num_services_generated >= self._parameters.num_services: def _schedule_request_setup(self) -> None: if self._generator.num_requests_generated >= self._parameters.num_requests: LOGGER.info('Generation Done!') #self._scheduler.shutdown() return iat = random.expovariate(1.0 / self._parameters.inter_arrival_time) run_date = datetime.utcnow() + timedelta(seconds=iat) self._scheduler.add_job( self._service_setup, trigger='date', run_date=run_date, timezone=pytz.utc) self._request_setup, trigger='date', run_date=run_date, timezone=pytz.utc) def _schedule_service_teardown(self, service : Dict) -> None: def _schedule_request_teardown(self, request : Dict) -> None: ht = random.expovariate(1.0 / self._parameters.holding_time) run_date = datetime.utcnow() + timedelta(seconds=ht) self._scheduler.add_job( self._service_teardown, args=(service,), trigger='date', run_date=run_date, timezone=pytz.utc) self._request_teardown, args=(request,), trigger='date', run_date=run_date, timezone=pytz.utc) def start(self): self._schedule_service_setup() self._schedule_request_setup() self._scheduler.start() def _service_setup(self) -> None: self._schedule_service_setup() def _request_setup(self) -> None: self._schedule_request_setup() service = self._service_generator.compose_service() if service is None: LOGGER.warning('No resources available to compose new service') request = self._generator.compose_request() if request is None: LOGGER.warning('No resources available to compose new request') return service_uuid = service['service_id']['service_uuid']['uuid'] src_device_uuid = service['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = service['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = service['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = service['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] if 'service_id' in request: service_uuid = request['service_id']['service_uuid']['uuid'] src_device_uuid = request['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Setup Service: uuid=%s src=%s:%s dst=%s:%s', service_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) if not self._parameters.dry_mode: service_add = copy.deepcopy(service) service_add['service_endpoint_ids'] = [] service_add['service_constraints'] = [] service_add['service_config'] = {'config_rules': []} request_add = copy.deepcopy(request) request_add['service_endpoint_ids'] = [] request_add['service_constraints'] = [] request_add['service_config'] = {'config_rules': []} service_client = ServiceClient() # create instances per request to load balance between pods service_client.CreateService(Service(**service_add)) service_client.UpdateService(Service(**service)) service_client.CreateService(Service(**request_add)) service_client.UpdateService(Service(**request)) service_client.close() self._schedule_service_teardown(service) elif 'slice_id' in request: slice_uuid = request['slice_id']['slice_uuid']['uuid'] src_device_uuid = request['slice_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['slice_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['slice_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['slice_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Setup Slice: uuid=%s src=%s:%s dst=%s:%s', slice_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) def _service_teardown(self, service : Dict) -> None: service_uuid = service['service_id']['service_uuid']['uuid'] src_device_uuid = service['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = service['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = service['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = service['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] if not self._parameters.dry_mode: request_add = copy.deepcopy(request) request_add['slice_endpoint_ids'] = [] request_add['slice_constraints'] = [] request_add['slice_config'] = {'config_rules': []} slice_client = SliceClient() # create instances per request to load balance between pods slice_client.CreateSlice(Slice(**request_add)) slice_client.UpdateSlice(Slice(**request)) slice_client.close() self._schedule_request_teardown(request) def _request_teardown(self, request : Dict) -> None: if 'service_id' in request: service_uuid = request['service_id']['service_uuid']['uuid'] src_device_uuid = request['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Teardown Service: uuid=%s src=%s:%s dst=%s:%s', service_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) if not self._parameters.dry_mode: service_client = ServiceClient() # create instances per request to load balance between pods service_client.DeleteService(ServiceId(**(service['service_id']))) service_client.DeleteService(ServiceId(**(request['service_id']))) service_client.close() elif 'slice_id' in request: slice_uuid = request['slice_id']['slice_uuid']['uuid'] src_device_uuid = request['slice_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['slice_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['slice_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['slice_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Teardown Slice: uuid=%s src=%s:%s dst=%s:%s', slice_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) if not self._parameters.dry_mode: slice_client = SliceClient() # create instances per request to load balance between pods slice_client.DeleteSlice(SliceId(**(request['slice_id']))) slice_client.close() self._service_generator.release_service(service) self._generator.release_request(request) src/tests/tools/load_gen/__main__.py +15 −11 Original line number Diff line number Diff line Loading @@ -13,10 +13,12 @@ # limitations under the License. import logging, sys from .Constants import SERVICE_TYPE_L2NM, SERVICE_TYPE_L3NM, SERVICE_TYPE_TAPI from .Constants import ( REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI, REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM) from .Parameters import Parameters from .ServiceGenerator import ServiceGenerator from .ServiceScheduler import ServiceScheduler from .RequestGenerator import RequestGenerator from .RequestScheduler import RequestScheduler logging.basicConfig(level=logging.INFO) LOGGER = logging.getLogger(__name__) Loading @@ -24,11 +26,13 @@ LOGGER = logging.getLogger(__name__) def main(): LOGGER.info('Starting...') parameters = Parameters( num_services = 100, service_types = [ SERVICE_TYPE_L2NM, SERVICE_TYPE_L3NM, #SERVICE_TYPE_TAPI, num_requests = 100, request_types = [ REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI, REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM, ], offered_load = 50, holding_time = 10, Loading @@ -36,11 +40,11 @@ def main(): ) LOGGER.info('Initializing Generator...') service_generator = ServiceGenerator(parameters) service_generator.initialize() generator = RequestGenerator(parameters) generator.initialize() LOGGER.info('Running Schedule...') scheduler = ServiceScheduler(parameters, service_generator) scheduler = RequestScheduler(parameters, generator) scheduler.start() LOGGER.info('Done!') Loading Loading
src/tests/tools/load_gen/Constants.py +5 −3 Original line number Diff line number Diff line Loading @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. SERVICE_TYPE_L2NM = 'l2nm' SERVICE_TYPE_L3NM = 'l3nm' SERVICE_TYPE_TAPI = 'tapi' REQUEST_TYPE_SERVICE_L2NM = 'svc-l2nm' REQUEST_TYPE_SERVICE_L3NM = 'svc-l3nm' REQUEST_TYPE_SERVICE_TAPI = 'svc-tapi' REQUEST_TYPE_SLICE_L2NM = 'slc-l2nm' REQUEST_TYPE_SLICE_L3NM = 'slc-l3nm'
src/tests/tools/load_gen/Parameters.py +5 −5 Original line number Diff line number Diff line Loading @@ -16,12 +16,12 @@ from typing import List, Optional class Parameters: def __init__( self, num_services : int, service_types : List[str], offered_load : Optional[float] = None, self, num_requests : int, request_types : List[str], offered_load : Optional[float] = None, inter_arrival_time : Optional[float] = None, holding_time : Optional[float] = None, dry_mode : bool = False ) -> None: self._num_services = num_services self._service_types = service_types self._num_requests = num_requests self._request_types = request_types self._offered_load = offered_load self._inter_arrival_time = inter_arrival_time self._holding_time = holding_time Loading @@ -35,10 +35,10 @@ class Parameters: self._holding_time = self._offered_load * self._inter_arrival_time @property def num_services(self): return self._num_services def num_requests(self): return self._num_requests @property def service_types(self): return self._service_types def request_types(self): return self._request_types @property def offered_load(self): return self._offered_load Loading
src/tests/tools/load_gen/ServiceGenerator.py→src/tests/tools/load_gen/RequestGenerator.py +139 −31 Original line number Diff line number Diff line Loading @@ -21,8 +21,11 @@ from common.tools.object_factory.Device import json_device_id from common.tools.object_factory.EndPoint import json_endpoint_id from common.tools.object_factory.Service import ( json_service_l2nm_planned, json_service_l3nm_planned, json_service_tapi_planned) from common.tools.object_factory.Slice import json_slice from context.client.ContextClient import ContextClient from .Constants import SERVICE_TYPE_L2NM, SERVICE_TYPE_L3NM, SERVICE_TYPE_TAPI from .Constants import ( REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI, REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM) from .Parameters import Parameters LOGGER = logging.getLogger(__name__) Loading @@ -32,11 +35,11 @@ ENDPOINT_COMPATIBILITY = { 'PHOTONIC_MEDIA:DWDM:G_50GHZ:INPUT' : 'PHOTONIC_MEDIA:DWDM:G_50GHZ:OUTPUT', } class ServiceGenerator: class RequestGenerator: def __init__(self, parameters : Parameters) -> None: self._parameters = parameters self._lock = threading.Lock() self._num_services = 0 self._num_requests = 0 self._available_device_endpoints : Dict[str, Set[str]] = dict() self._used_device_endpoints : Dict[str, Dict[str, str]] = dict() self._endpoint_ids_to_types : Dict[Tuple[str, str], str] = dict() Loading Loading @@ -79,7 +82,7 @@ class ServiceGenerator: endpoints_for_type.discard(endpoint_key) @property def num_services_generated(self): return self._num_services def num_requests_generated(self): return self._num_requests def dump_state(self) -> None: with self._lock: Loading Loading @@ -139,19 +142,26 @@ class ServiceGenerator: self._used_device_endpoints.setdefault(device_uuid, set()).pop(endpoint_uuid, None) self._available_device_endpoints.setdefault(device_uuid, set()).add(endpoint_uuid) def compose_service(self) -> Optional[Dict]: def compose_request(self) -> Optional[Dict]: with self._lock: self._num_services += 1 num_service = self._num_services #service_uuid = str(uuid.uuid4()) service_uuid = 'svc_{:d}'.format(num_service) self._num_requests += 1 num_request = self._num_requests # choose service type service_type = random.choice(self._parameters.service_types) #request_uuid = str(uuid.uuid4()) request_uuid = 'svc_{:d}'.format(num_request) # choose request type request_type = random.choice(self._parameters.request_types) if request_type in {REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI}: return self._compose_service(num_request, request_uuid, request_type) elif request_type in {REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM}: return self._compose_slice(num_request, request_uuid, request_type) def _compose_service(self, num_request : int, request_uuid : str, request_type : str) -> Optional[Dict]: # choose source endpoint src_endpoint_types = set(ENDPOINT_COMPATIBILITY.keys()) if service_type in {SERVICE_TYPE_TAPI} else None src = self._use_device_endpoint(service_uuid, endpoint_types=src_endpoint_types) src_endpoint_types = set(ENDPOINT_COMPATIBILITY.keys()) if request_type in {REQUEST_TYPE_SERVICE_TAPI} else None src = self._use_device_endpoint(request_uuid, endpoint_types=src_endpoint_types) if src is None: LOGGER.warning('>> No source endpoint is available') return None Loading @@ -160,14 +170,14 @@ class ServiceGenerator: # identify compatible destination endpoint types src_endpoint_type = self._endpoint_ids_to_types.get((src_device_uuid,src_endpoint_uuid)) dst_endpoint_type = ENDPOINT_COMPATIBILITY.get(src_endpoint_type) dst_endpoint_types = {dst_endpoint_type} if service_type in {SERVICE_TYPE_TAPI} else None dst_endpoint_types = {dst_endpoint_type} if request_type in {REQUEST_TYPE_SERVICE_TAPI} else None # identify expluded destination devices exclude_device_uuids = {} if service_type in {SERVICE_TYPE_TAPI} else {src_device_uuid} # identify excluded destination devices exclude_device_uuids = {} if request_type in {REQUEST_TYPE_SERVICE_TAPI} else {src_device_uuid} # choose feasible destination endpoint dst = self._use_device_endpoint( service_uuid, endpoint_types=dst_endpoint_types, exclude_device_uuids=exclude_device_uuids) request_uuid, endpoint_types=dst_endpoint_types, exclude_device_uuids=exclude_device_uuids) # if destination endpoint not found, release source, and terminate current service generation if dst is None: Loading @@ -182,12 +192,12 @@ class ServiceGenerator: json_endpoint_id(json_device_id(dst_device_uuid), dst_endpoint_uuid), ] if service_type == SERVICE_TYPE_L2NM: if request_type == REQUEST_TYPE_SERVICE_L2NM: constraints = [ json_constraint_custom('bandwidth[gbps]', '10.0'), json_constraint_custom('latency[ms]', '20.0'), ] vlan_id = num_service % 1000 vlan_id = num_request % 1000 circuit_id = '{:03d}'.format(vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) dst_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) Loading @@ -211,15 +221,15 @@ class ServiceGenerator: }), ] return json_service_l2nm_planned( service_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) request_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) elif service_type == SERVICE_TYPE_L3NM: elif request_type == REQUEST_TYPE_SERVICE_L3NM: constraints = [ json_constraint_custom('bandwidth[gbps]', '10.0'), json_constraint_custom('latency[ms]', '20.0'), ] vlan_id = num_service % 1000 bgp_as = 60000 + (num_service % 10000) vlan_id = num_request % 1000 bgp_as = 60000 + (num_request % 10000) bgp_route_target = '{:5d}:{:03d}'.format(bgp_as, 333) route_distinguisher = '{:5d}:{:03d}'.format(bgp_as, vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) Loading Loading @@ -250,9 +260,9 @@ class ServiceGenerator: }), ] return json_service_l3nm_planned( service_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) request_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) elif service_type == SERVICE_TYPE_TAPI: elif request_type == REQUEST_TYPE_SERVICE_TAPI: config_rules = [ json_config_rule_set('/settings', { 'capacity_value' : 50.0, Loading @@ -263,10 +273,108 @@ class ServiceGenerator: }), ] return json_service_tapi_planned( service_uuid, endpoint_ids=endpoint_ids, constraints=[], config_rules=config_rules) request_uuid, endpoint_ids=endpoint_ids, constraints=[], config_rules=config_rules) def _compose_slice(self, num_request : int, request_uuid : str, request_type : str) -> Optional[Dict]: # choose source endpoint src = self._use_device_endpoint(request_uuid) if src is None: LOGGER.warning('>> No source endpoint is available') return None src_device_uuid,src_endpoint_uuid = src # identify excluded destination devices exclude_device_uuids = {} if request_type in {REQUEST_TYPE_SERVICE_TAPI} else {src_device_uuid} # choose feasible destination endpoint dst = self._use_device_endpoint(request_uuid, exclude_device_uuids=exclude_device_uuids) def release_service(self, json_service : Dict) -> None: for endpoint_id in json_service['service_endpoint_ids']: # if destination endpoint not found, release source, and terminate current service generation if dst is None: LOGGER.warning('>> No destination endpoint is available') self._release_device_endpoint(src_device_uuid, src_endpoint_uuid) return None # compose endpoints dst_device_uuid,dst_endpoint_uuid = dst endpoint_ids = [ json_endpoint_id(json_device_id(src_device_uuid), src_endpoint_uuid), json_endpoint_id(json_device_id(dst_device_uuid), dst_endpoint_uuid), ] constraints = [ json_constraint_custom('bandwidth[gbps]', '10.0'), json_constraint_custom('latency[ms]', '20.0'), ] if request_type == REQUEST_TYPE_SLICE_L2NM: vlan_id = num_request % 1000 circuit_id = '{:03d}'.format(vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) dst_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) config_rules = [ json_config_rule_set('/settings', { 'mtu': 1512 }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(src_device_uuid, src_endpoint_uuid), { 'router_id': src_router_id, 'sub_interface_index': vlan_id, 'vlan_id': vlan_id, 'remote_router': dst_router_id, 'circuit_id': circuit_id, }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(dst_device_uuid, dst_endpoint_uuid), { 'router_id': dst_router_id, 'sub_interface_index': vlan_id, 'vlan_id': vlan_id, 'remote_router': src_router_id, 'circuit_id': circuit_id, }), ] elif request_type == REQUEST_TYPE_SLICE_L3NM: vlan_id = num_request % 1000 bgp_as = 60000 + (num_request % 10000) bgp_route_target = '{:5d}:{:03d}'.format(bgp_as, 333) route_distinguisher = '{:5d}:{:03d}'.format(bgp_as, vlan_id) src_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) dst_router_id = '10.0.0.{:d}'.format(int(src_device_uuid.replace('R', ''))) src_address_ip = '.'.join([src_device_uuid.replace('R', ''), '0'] + src_endpoint_uuid.split('/')) dst_address_ip = '.'.join([dst_device_uuid.replace('R', ''), '0'] + dst_endpoint_uuid.split('/')) config_rules = [ json_config_rule_set('/settings', { 'mtu' : 1512, 'bgp_as' : bgp_as, 'bgp_route_target': bgp_route_target, }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(src_device_uuid, src_endpoint_uuid), { 'router_id' : src_router_id, 'route_distinguisher': route_distinguisher, 'sub_interface_index': vlan_id, 'vlan_id' : vlan_id, 'address_ip' : src_address_ip, 'address_prefix' : 16, }), json_config_rule_set('/device[{:s}]/endpoint[{:s}]/settings'.format(dst_device_uuid, dst_endpoint_uuid), { 'router_id' : dst_router_id, 'route_distinguisher': route_distinguisher, 'sub_interface_index': vlan_id, 'vlan_id' : vlan_id, 'address_ip' : dst_address_ip, 'address_prefix' : 16, }), ] return json_slice( request_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules) def release_request(self, json_request : Dict) -> None: if 'service_id' in json_request: for endpoint_id in json_request['service_endpoint_ids']: device_uuid = endpoint_id['device_id']['device_uuid']['uuid'] endpoint_uuid = endpoint_id['endpoint_uuid']['uuid'] self._release_device_endpoint(device_uuid, endpoint_uuid) elif 'slice_id' in json_request: for endpoint_id in json_request['slice_endpoint_ids']: device_uuid = endpoint_id['device_id']['device_uuid']['uuid'] endpoint_uuid = endpoint_id['endpoint_uuid']['uuid'] self._release_device_endpoint(device_uuid, endpoint_uuid)
src/tests/tools/load_gen/ServiceScheduler.py→src/tests/tools/load_gen/RequestScheduler.py +144 −0 Original line number Diff line number Diff line Loading @@ -18,18 +18,19 @@ from apscheduler.executors.pool import ThreadPoolExecutor from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.schedulers.blocking import BlockingScheduler from typing import Dict from common.proto.context_pb2 import Service, ServiceId from common.proto.context_pb2 import Service, ServiceId, Slice, SliceId from service.client.ServiceClient import ServiceClient from slice.client.SliceClient import SliceClient from .Parameters import Parameters from .ServiceGenerator import ServiceGenerator from .RequestGenerator import RequestGenerator logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING) logging.getLogger('apscheduler.scheduler').setLevel(logging.WARNING) LOGGER = logging.getLogger(__name__) class ServiceScheduler: def __init__(self, parameters : Parameters, service_generator : ServiceGenerator) -> None: class RequestScheduler: def __init__(self, parameters : Parameters, generator : RequestGenerator) -> None: self._scheduler = BlockingScheduler() self._scheduler.configure( jobstores = {'default': MemoryJobStore()}, Loading @@ -41,66 +42,103 @@ class ServiceScheduler: }, timezone=pytz.utc) self._parameters = parameters self._service_generator = service_generator self._generator = generator def _schedule_service_setup(self) -> None: if self._service_generator.num_services_generated >= self._parameters.num_services: def _schedule_request_setup(self) -> None: if self._generator.num_requests_generated >= self._parameters.num_requests: LOGGER.info('Generation Done!') #self._scheduler.shutdown() return iat = random.expovariate(1.0 / self._parameters.inter_arrival_time) run_date = datetime.utcnow() + timedelta(seconds=iat) self._scheduler.add_job( self._service_setup, trigger='date', run_date=run_date, timezone=pytz.utc) self._request_setup, trigger='date', run_date=run_date, timezone=pytz.utc) def _schedule_service_teardown(self, service : Dict) -> None: def _schedule_request_teardown(self, request : Dict) -> None: ht = random.expovariate(1.0 / self._parameters.holding_time) run_date = datetime.utcnow() + timedelta(seconds=ht) self._scheduler.add_job( self._service_teardown, args=(service,), trigger='date', run_date=run_date, timezone=pytz.utc) self._request_teardown, args=(request,), trigger='date', run_date=run_date, timezone=pytz.utc) def start(self): self._schedule_service_setup() self._schedule_request_setup() self._scheduler.start() def _service_setup(self) -> None: self._schedule_service_setup() def _request_setup(self) -> None: self._schedule_request_setup() service = self._service_generator.compose_service() if service is None: LOGGER.warning('No resources available to compose new service') request = self._generator.compose_request() if request is None: LOGGER.warning('No resources available to compose new request') return service_uuid = service['service_id']['service_uuid']['uuid'] src_device_uuid = service['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = service['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = service['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = service['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] if 'service_id' in request: service_uuid = request['service_id']['service_uuid']['uuid'] src_device_uuid = request['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Setup Service: uuid=%s src=%s:%s dst=%s:%s', service_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) if not self._parameters.dry_mode: service_add = copy.deepcopy(service) service_add['service_endpoint_ids'] = [] service_add['service_constraints'] = [] service_add['service_config'] = {'config_rules': []} request_add = copy.deepcopy(request) request_add['service_endpoint_ids'] = [] request_add['service_constraints'] = [] request_add['service_config'] = {'config_rules': []} service_client = ServiceClient() # create instances per request to load balance between pods service_client.CreateService(Service(**service_add)) service_client.UpdateService(Service(**service)) service_client.CreateService(Service(**request_add)) service_client.UpdateService(Service(**request)) service_client.close() self._schedule_service_teardown(service) elif 'slice_id' in request: slice_uuid = request['slice_id']['slice_uuid']['uuid'] src_device_uuid = request['slice_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['slice_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['slice_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['slice_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Setup Slice: uuid=%s src=%s:%s dst=%s:%s', slice_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) def _service_teardown(self, service : Dict) -> None: service_uuid = service['service_id']['service_uuid']['uuid'] src_device_uuid = service['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = service['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = service['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = service['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] if not self._parameters.dry_mode: request_add = copy.deepcopy(request) request_add['slice_endpoint_ids'] = [] request_add['slice_constraints'] = [] request_add['slice_config'] = {'config_rules': []} slice_client = SliceClient() # create instances per request to load balance between pods slice_client.CreateSlice(Slice(**request_add)) slice_client.UpdateSlice(Slice(**request)) slice_client.close() self._schedule_request_teardown(request) def _request_teardown(self, request : Dict) -> None: if 'service_id' in request: service_uuid = request['service_id']['service_uuid']['uuid'] src_device_uuid = request['service_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['service_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['service_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['service_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Teardown Service: uuid=%s src=%s:%s dst=%s:%s', service_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) if not self._parameters.dry_mode: service_client = ServiceClient() # create instances per request to load balance between pods service_client.DeleteService(ServiceId(**(service['service_id']))) service_client.DeleteService(ServiceId(**(request['service_id']))) service_client.close() elif 'slice_id' in request: slice_uuid = request['slice_id']['slice_uuid']['uuid'] src_device_uuid = request['slice_endpoint_ids'][0]['device_id']['device_uuid']['uuid'] src_endpoint_uuid = request['slice_endpoint_ids'][0]['endpoint_uuid']['uuid'] dst_device_uuid = request['slice_endpoint_ids'][1]['device_id']['device_uuid']['uuid'] dst_endpoint_uuid = request['slice_endpoint_ids'][1]['endpoint_uuid']['uuid'] LOGGER.info('Teardown Slice: uuid=%s src=%s:%s dst=%s:%s', slice_uuid, src_device_uuid, src_endpoint_uuid, dst_device_uuid, dst_endpoint_uuid) if not self._parameters.dry_mode: slice_client = SliceClient() # create instances per request to load balance between pods slice_client.DeleteSlice(SliceId(**(request['slice_id']))) slice_client.close() self._service_generator.release_service(service) self._generator.release_request(request)
src/tests/tools/load_gen/__main__.py +15 −11 Original line number Diff line number Diff line Loading @@ -13,10 +13,12 @@ # limitations under the License. import logging, sys from .Constants import SERVICE_TYPE_L2NM, SERVICE_TYPE_L3NM, SERVICE_TYPE_TAPI from .Constants import ( REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI, REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM) from .Parameters import Parameters from .ServiceGenerator import ServiceGenerator from .ServiceScheduler import ServiceScheduler from .RequestGenerator import RequestGenerator from .RequestScheduler import RequestScheduler logging.basicConfig(level=logging.INFO) LOGGER = logging.getLogger(__name__) Loading @@ -24,11 +26,13 @@ LOGGER = logging.getLogger(__name__) def main(): LOGGER.info('Starting...') parameters = Parameters( num_services = 100, service_types = [ SERVICE_TYPE_L2NM, SERVICE_TYPE_L3NM, #SERVICE_TYPE_TAPI, num_requests = 100, request_types = [ REQUEST_TYPE_SERVICE_L2NM, REQUEST_TYPE_SERVICE_L3NM, REQUEST_TYPE_SERVICE_TAPI, REQUEST_TYPE_SLICE_L2NM, REQUEST_TYPE_SLICE_L3NM, ], offered_load = 50, holding_time = 10, Loading @@ -36,11 +40,11 @@ def main(): ) LOGGER.info('Initializing Generator...') service_generator = ServiceGenerator(parameters) service_generator.initialize() generator = RequestGenerator(parameters) generator.initialize() LOGGER.info('Running Schedule...') scheduler = ServiceScheduler(parameters, service_generator) scheduler = RequestScheduler(parameters, generator) scheduler.start() LOGGER.info('Done!') Loading