ComposeConfigRules.py

# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
#
# 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 itertools, json, re
from typing import Dict, List, Optional, Tuple
from common.proto.context_pb2 import ConfigRule
from common.tools.object_factory.ConfigRule import json_config_rule_set

SETTINGS_RULE_NAME = '/settings'

DEV_EP_SETTINGS = re.compile(r'\/device\[([^\]]+)\]\/endpoint\[([^\]]+)\]\/settings')

L2NM_SETTINGS_FIELD_DEFAULTS = {
    'encapsulation_type': 'dot1q',
    'vlan_id'           : 100,
    'mtu'               : 1450,
}

L3NM_SETTINGS_FIELD_DEFAULTS = {
    'encapsulation_type': 'dot1q',
    'vlan_id'           : 100,
    'mtu'               : 1450,
}

TAPI_SETTINGS_FIELD_DEFAULTS = {
    'capacity_value'  : 50.0,
    'capacity_unit'   : 'GHz',
    'layer_proto_name': 'PHOTONIC_MEDIA',
    'layer_proto_qual': 'tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_NMC',
    'direction'       : 'UNIDIRECTIONAL',
}

def find_custom_config_rule(config_rules : List, resource_name : str) -> Optional[Dict]:
    resource_value : Optional[Dict] = None
    for config_rule in config_rules:
        if config_rule.WhichOneof('config_rule') != 'custom': continue
        if config_rule.custom.resource_key != resource_name: continue
        resource_value = json.loads(config_rule.custom.resource_value)
    return resource_value

def compose_config_rules(
    main_service_config_rules : List, subservice_config_rules : List, field_defaults : Dict
) -> None:
    settings = find_custom_config_rule(main_service_config_rules, SETTINGS_RULE_NAME)
    if settings is None: return

    json_settings = {}
    for field_name,default_value in field_defaults.items():
        json_settings[field_name] = settings.get(field_name, default_value)

    config_rule = ConfigRule(**json_config_rule_set('/settings', json_settings))
    subservice_config_rules.append(config_rule)

def compose_l2nm_config_rules(main_service_config_rules : List, subservice_config_rules : List) -> None:
    compose_config_rules(main_service_config_rules, subservice_config_rules, L2NM_SETTINGS_FIELD_DEFAULTS)

def compose_l3nm_config_rules(main_service_config_rules : List, subservice_config_rules : List) -> None:
    compose_config_rules(main_service_config_rules, subservice_config_rules, L3NM_SETTINGS_FIELD_DEFAULTS)

def compose_tapi_config_rules(main_service_config_rules : List, subservice_config_rules : List) -> None:
    compose_config_rules(main_service_config_rules, subservice_config_rules, TAPI_SETTINGS_FIELD_DEFAULTS)

def compose_device_config_rules(
    config_rules : List, subservice_config_rules : List, path_hops : List,
    device_name_mapping : Dict[str, str], endpoint_name_mapping : Dict[Tuple[str, str], str]
) -> None:

    endpoints_traversed = set()
    for path_hop in path_hops:
        device_uuid_or_name = path_hop['device']
        endpoints_traversed.add((device_uuid_or_name, path_hop['ingress_ep']))
        endpoints_traversed.add((device_uuid_or_name, path_hop['egress_ep']))

    for config_rule in config_rules:
        if config_rule.WhichOneof('config_rule') != 'custom': continue
        match = DEV_EP_SETTINGS.match(config_rule.custom.resource_key)
        if match is None: continue

        device_uuid_or_name = match.group(1)
        device_name_or_uuid = device_name_mapping[device_uuid_or_name]
        device_keys = {device_uuid_or_name, device_name_or_uuid}

        endpoint_uuid_or_name = match.group(2)
        endpoint_name_or_uuid_1 = endpoint_name_mapping[(device_uuid_or_name, endpoint_uuid_or_name)]
        endpoint_name_or_uuid_2 = endpoint_name_mapping[(device_name_or_uuid, endpoint_uuid_or_name)]
        endpoint_keys = {endpoint_uuid_or_name, endpoint_name_or_uuid_1, endpoint_name_or_uuid_2}

        device_endpoint_keys = set(itertools.product(device_keys, endpoint_keys))
        if len(device_endpoint_keys.intersection(endpoints_traversed)) == 0: continue
        subservice_config_rules.append(config_rule)