Tools - Load Generator:

# 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.

from typing import Optional

class Parameters:

    def __init__(

        self, num_services : int, 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._offered_load = offered_load

        self._inter_arrival_time = inter_arrival_time

        self._holding_time = holding_time

        self._dry_mode = dry_mode

        if self._offered_load is None and self._holding_time is not None and self._inter_arrival_time is not None:

            self._offered_load = self._holding_time / self._inter_arrival_time

        elif self._offered_load is not None and self._holding_time is not None and self._inter_arrival_time is None:

            self._inter_arrival_time = self._holding_time / self._offered_load

        elif self._offered_load is not None and self._holding_time is None and self._inter_arrival_time is not None:

            self._holding_time = self._offered_load * self._inter_arrival_time

    @property

    def num_services(self): return self._num_services

    @property

    def offered_load(self): return self._offered_load

    @property

    def inter_arrival_time(self): return self._inter_arrival_time

    @property

    def holding_time(self): return self._holding_time

    @property

    def dry_mode(self): return self._dry_mode

# 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 logging, json, random, threading

from typing import Dict, Optional, Set, Tuple

from common.proto.context_pb2 import Empty

from common.tools.object_factory.Constraint import json_constraint_custom

from common.tools.object_factory.ConfigRule import json_config_rule_set

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

from context.client.ContextClient import ContextClient

LOGGER = logging.getLogger(__name__)

class ServiceGenerator:

    def __init__(self) -> None:

        self._lock = threading.Lock()

        self._num_services = 0

        self._available_device_endpoints : Dict[str, Set[str]] = dict()

        self._used_device_endpoints : Dict[str, Dict[str, str]] = dict()

    def initialize(self) -> None:

        with self._lock:

            self._available_device_endpoints.clear()

            self._used_device_endpoints.clear()

            context_client = ContextClient()

            devices = context_client.ListDevices(Empty())

            for device in devices.devices:

                device_uuid = device.device_id.device_uuid.uuid

                _endpoints = self._available_device_endpoints.setdefault(device_uuid, set())

                for endpoint in device.device_endpoints:

                    endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid

                    _endpoints.add(endpoint_uuid)

            links = context_client.ListLinks(Empty())

            for link in links.links:

                for endpoint_id in link.link_endpoint_ids:

                    device_uuid = endpoint_id.device_id.device_uuid.uuid

                    endpoint_uuid = endpoint_id.endpoint_uuid.uuid

                    _endpoints = self._available_device_endpoints.get(device_uuid, set())

                    _endpoints.discard(endpoint_uuid)

                    if len(_endpoints) == 0: self._available_device_endpoints.pop(device_uuid, None)

    @property

    def num_services_generated(self): return self._num_services

    def dump_state(self) -> None:

        with self._lock:

            _endpoints = {

                device_uuid:[endpoint_uuid for endpoint_uuid in endpoint_uuids]

                for device_uuid,endpoint_uuids in self._available_device_endpoints.items()

            }

            LOGGER.info('[dump_state] available_device_endpoints = {:s}'.format(json.dumps(_endpoints)))

            LOGGER.info('[dump_state] used_device_endpoints = {:s}'.format(json.dumps(self._used_device_endpoints)))

    def _use_device_endpoint(

        self, service_uuid : str, exclude_device_uuids : Set[str] = set()

    ) -> Optional[Tuple[str, str]]:

        with self._lock:

            elegible_device_endpoints = {

                device_uuid:device_endpoint_uuids

                for device_uuid,device_endpoint_uuids in self._available_device_endpoints.items()

                if device_uuid not in exclude_device_uuids and len(device_endpoint_uuids) > 0

            }

            if len(elegible_device_endpoints) == 0: return None

            device_uuid = random.choice(list(elegible_device_endpoints.keys()))

            device_endpoint_uuids = elegible_device_endpoints.get(device_uuid)

            endpoint_uuid = random.choice(list(device_endpoint_uuids))

            self._available_device_endpoints.setdefault(device_uuid, set()).discard(endpoint_uuid)

            self._used_device_endpoints.setdefault(device_uuid, dict())[endpoint_uuid] = service_uuid

            return device_uuid, endpoint_uuid

    def _release_device_endpoint(self, device_uuid : str, endpoint_uuid : str) -> None:

        with self._lock:

            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]:

        with self._lock:

            self._num_services += 1

            num_service = self._num_services

        #service_uuid = str(uuid.uuid4())

        service_uuid = 'svc_{:d}'.format(num_service)

        src = self._use_device_endpoint(service_uuid)

        if src is None: return None

        src_device_uuid,src_endpoint_uuid = src

        dst = self._use_device_endpoint(service_uuid, exclude_device_uuids={src_device_uuid})

        if dst is None:

            self._release_device_endpoint(src_device_uuid, src_endpoint_uuid)

            return None

        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'),

        ]

        vlan_id = num_service % 1000

        circuit_id = '{:03d}'.format(vlan_id)

        src_router_id = '.'.join([src_device_uuid.replace('R', ''), '0'] + src_endpoint_uuid.split('/'))

        dst_router_id = '.'.join([dst_device_uuid.replace('R', ''), '0'] + dst_endpoint_uuid.split('/'))

        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,

            }),

        ]

        return json_service_l2nm_planned(

            service_uuid, endpoint_ids=endpoint_ids, constraints=constraints, config_rules=config_rules)

    def release_service(self, json_service : Dict) -> None:

        for endpoint_id in json_service['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)

# 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 copy, logging, pytz, random

from datetime import datetime, timedelta

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 service.client.ServiceClient import ServiceClient

from .Parameters import Parameters

from .ServiceGenerator import ServiceGenerator

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:

        self._scheduler = BlockingScheduler()

        self._scheduler.configure(

            jobstores = {'default': MemoryJobStore()},

            executors = {'default': ThreadPoolExecutor(max_workers=10)},

            job_defaults = {

                'coalesce': False,

                'max_instances': 100,

                'misfire_grace_time': 120,

            },

            timezone=pytz.utc)

        self._parameters = parameters

        self._service_generator = service_generator

    def _schedule_service_setup(self) -> None:

        if self._service_generator.num_services_generated >= self._parameters.num_services:

            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)

    def _schedule_service_teardown(self, service : 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)

    def start(self):

        self._schedule_service_setup()

        self._scheduler.start()

    def _service_setup(self) -> None:

        self._schedule_service_setup()

        service = self._service_generator.compose_service()

        if service is None:

            LOGGER.warning('No resources available to compose new service')

            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']

        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': []}

            service_client = ServiceClient()    # create instances per request to load balance between pods

            service_client.CreateService(Service(**service_add))

            service_client.UpdateService(Service(**service))

        self._schedule_service_teardown(service)

    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']

        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'])))

        self._service_generator.release_service(service)

# 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.

# 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 logging, sys

from .Parameters import Parameters

from .ServiceGenerator import ServiceGenerator

from .ServiceScheduler import ServiceScheduler

logging.basicConfig(level=logging.INFO)

LOGGER = logging.getLogger(__name__)

def main():

    LOGGER.info('Starting...')

    parameters = Parameters(

        num_services = 100,

        offered_load = 50,

        holding_time = 10,

        dry_mode     = False, # in dry mode, no request is sent to TeraFlowSDN

    )

    LOGGER.info('Initializing Generator...')

    service_generator = ServiceGenerator()

    service_generator.initialize()

    LOGGER.info('Running Schedule...')

    scheduler = ServiceScheduler(parameters, service_generator)

    scheduler.start()

    LOGGER.info('Done!')

    return 0

if __name__ == '__main__':

    sys.exit(main())