EthtServiceHandler.py

# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (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 enum, logging
from typing import Dict, List, Optional, Tuple, Union
from .RestApiClient import HTTP_STATUS_CREATED, HTTP_STATUS_NO_CONTENT, HTTP_STATUS_OK, RestApiClient

LOGGER = logging.getLogger(__name__)

class BandwidthProfileTypeEnum(enum.Enum):
    MEF_10_BWP = 'ietf-eth-tran-types:mef-10-bwp'

class EndpointLayerSpecificAccessTypeEnum(enum.Enum):
    PORT = 'port'

class EndpointProtectionRoleEnum(enum.Enum):
    WORK = 'work'

class OptimizationMetricRole(enum.Enum):
    WORK = 'work'

class OptimizationMetricType(enum.Enum):
    PATH_METRIC_TE = 'ietf-te-types:path-metric-te'

class OuterTagTypeEnum(enum.Enum):
    CLASSIFY_C_VLAN = 'ietf-eth-tran-types:classify-c-vlan'

class ServiceClassificationTypeEnum(enum.Enum):
    VLAN_CLASSIFICATION = 'ietf-eth-tran-type:vlan-classification'

class ServiceTypeEnum(enum.Enum):
    MP2MP = 'op-mp2mp-svc'
    P2MP  = 'op-p2mp-svc'

def compose_outer_tag(tag_type : OuterTagTypeEnum, vlan_value : int) -> Dict:
    return {'tag-type': tag_type.value, 'vlan-value': vlan_value}

def compose_ingress_egress_bandwidth_profile() -> Dict:
    return {
        'bandwidth-profile-type': BandwidthProfileTypeEnum.MEF_10_BWP.value,
        'CIR': 10_000_000,
        'EIR': 10_000_000,
    }

def compose_layer_specific_access_type() -> Dict:
    return {'access-type': EndpointLayerSpecificAccessTypeEnum.PORT.value}

def compose_static_route(prefix : str, mask : int, next_hop : str) -> Dict:
    return {'destination': prefix, 'destination-mask': mask, 'next-hop': next_hop}

def compose_static_route_list(static_routes : List[Tuple[str, int, str]]) -> List[Dict]:
    return [
        compose_static_route(prefix, mask, next_hop)
        for prefix, mask, next_hop in static_routes
    ]

def compose_etht_service_endpoint(
    node_id : str, tp_id : str, vlan_value : int, static_routes : List[Tuple[str, int, str]] = list()
) -> Dict:
    return {
        'node-id'           : node_id,
        'tp-id'             : tp_id,
        'protection-role'   : EndpointProtectionRoleEnum.WORK.value,
        'layer-specific'    : compose_layer_specific_access_type(),
        'is-extendable'     : False,
        'is-terminal'       : True,
        'static-route-list' : compose_static_route_list(static_routes),
        'outer-tag'         : compose_outer_tag(OuterTagTypeEnum.CLASSIFY_C_VLAN, vlan_value),
        'service-classification-type'     : ServiceClassificationTypeEnum.VLAN_CLASSIFICATION.value,
        'ingress-egress-bandwidth-profile': compose_ingress_egress_bandwidth_profile(),
    }

def compose_optimizations() -> Dict:
    return {'optimization-metric': [{
        'metric-role': OptimizationMetricRole.WORK.value,
        'metric-type': OptimizationMetricType.PATH_METRIC_TE.value,
    }]}

def compose_etht_service(
    name : str, service_type : ServiceTypeEnum, osu_tunnel_name : str,
    src_node_id : str, src_tp_id : str, src_vlan_tag : int, dst_node_id : str, dst_tp_id : str, dst_vlan_tag : int,
    src_static_routes : List[Tuple[str, int, str]] = list(), dst_static_routes : List[Tuple[str, int, str]] = list()
) -> Dict:
    return {'ietf-eth-tran-service:etht-svc': {'etht-svc-instances': [{
        'etht-svc-name' : name,
        'etht-svc-title': name.upper(),
        'etht-svc-type' : service_type.value,
        'source-endpoints': {'source-endpoint': [
            compose_etht_service_endpoint(src_node_id, src_tp_id, src_vlan_tag, src_static_routes),
        ]},
        'destination-endpoints': {'destination-endpoint': [
            compose_etht_service_endpoint(dst_node_id, dst_tp_id, dst_vlan_tag, dst_static_routes),
        ]},
        'svc-tunnel': [{'tunnel-name': osu_tunnel_name}],
        'optimizations': compose_optimizations(),
    }]}}

class EthtServiceHandler:
    def __init__(self, rest_api_client : RestApiClient) -> None:
        self._rest_api_client = rest_api_client
        self._object_name  = 'EthtService'
        self._subpath_root = '/ietf-eth-tran-service:etht-svc'
        self._subpath_item = self._subpath_root + '/etht-svc-instances="{etht_service_name:s}"'

    def _rest_api_get(self, etht_service_name : Optional[str] = None) -> Union[Dict, List]:
        if etht_service_name is None:
            subpath_url = self._subpath_root
        else:
            subpath_url = self._subpath_item.format(etht_service_name=etht_service_name)
        return self._rest_api_client.get(
            self._object_name, subpath_url, expected_http_status={HTTP_STATUS_OK}
        )

    def _rest_api_update(self, data : Dict) -> bool:
        return self._rest_api_client.update(
            self._object_name, self._subpath_root, data, expected_http_status={HTTP_STATUS_CREATED}
        )

    def _rest_api_delete(self, etht_service_name : str) -> bool:
        if etht_service_name is None: raise Exception('etht_service_name is None')
        subpath_url = self._subpath_item.format(etht_service_name=etht_service_name)
        return self._rest_api_client.delete(
            self._object_name, subpath_url, expected_http_status={HTTP_STATUS_NO_CONTENT}
        )

    def get(self, etht_service_name : Optional[str] = None) -> Union[Dict, List]:
        data = self._rest_api_get(etht_service_name=etht_service_name)

        if not isinstance(data, dict): raise ValueError('data should be a dict')
        if 'ietf-eth-tran-service:etht-svc' not in data:
            raise ValueError('data does not contain key "ietf-eth-tran-service:etht-svc"')
        data = data['ietf-eth-tran-service:etht-svc']
        if 'etht-svc-instances' not in data:
            raise ValueError('data["ietf-eth-tran-service:etht-svc"] does not contain key "etht-svc-instances"')
        data = data['etht-svc-instances']
        if not isinstance(data, list):
            raise ValueError('data["ietf-eth-tran-service:etht-svc"]["etht-svc-instances"] should be a list')

        etht_services : List[Dict] = list()
        for item in data:
            src_endpoints = item['source-endpoints']['source-endpoint']
            if len(src_endpoints) != 1:
                MSG = 'EthtService({:s}) has zero/multiple source endpoints'
                raise Exception(MSG.format(str(item)))
            src_endpoint = src_endpoints[0]

            dst_endpoints = item['destination-endpoints']['destination-endpoint']
            if len(dst_endpoints) != 1:
                MSG = 'EthtService({:s}) has zero/multiple destination endpoints'
                raise Exception(MSG.format(str(item)))
            dst_endpoint = dst_endpoints[0]

            svc_tunnels = item['svc-tunnel']
            if len(svc_tunnels) != 1:
                MSG = 'EthtService({:s}) has zero/multiple service tunnels'
                raise Exception(MSG.format(str(item)))
            svc_tunnel = svc_tunnels[0]

            etht_service = {
                'name'             : item['etht-svc-name'],
                'service_type'     : item['etht-svc-type'],
                'osu_tunnel_name'  : svc_tunnel['tunnel-name'],

                'src_node_id'      : src_endpoint['node-id'],
                'src_tp_id'        : src_endpoint['tp-id'],
                'src_vlan_tag'     : src_endpoint['outer-tag']['vlan-value'],
                'src_static_routes': [
                    [static_route['destination'], static_route['destination-mask'], static_route['next-hop']]
                    for static_route in src_endpoint.get('static-route-list', list())
                ],

                'dst_node_id'      : dst_endpoint['node-id'],
                'dst_tp_id'        : dst_endpoint['tp-id'],
                'dst_vlan_tag'     : dst_endpoint['outer-tag']['vlan-value'],
                'dst_static_routes': [
                    [static_route['destination'], static_route['destination-mask'], static_route['next-hop']]
                    for static_route in dst_endpoint.get('static-route-list', list())
                ],
            }
            etht_services.append(etht_service)

        return etht_services

    def update(self, parameters : Dict) -> bool:
        name              = parameters['name'           ]
        service_type      = parameters['service_type'   ]
        osu_tunnel_name   = parameters['osu_tunnel_name']

        src_node_id       = parameters['src_node_id'    ]
        src_tp_id         = parameters['src_tp_id'      ]
        src_vlan_tag      = parameters['src_vlan_tag'   ]
        src_static_routes = parameters.get('src_static_routes', [])

        dst_node_id       = parameters['dst_node_id'    ]
        dst_tp_id         = parameters['dst_tp_id'      ]
        dst_vlan_tag      = parameters['dst_vlan_tag'   ]
        dst_static_routes = parameters.get('dst_static_routes', [])

        service_type = ServiceTypeEnum._value2member_map_[service_type]

        data = compose_etht_service(
            name, service_type, osu_tunnel_name,
            src_node_id, src_tp_id, src_vlan_tag, dst_node_id, dst_tp_id, dst_vlan_tag,
            src_static_routes=src_static_routes, dst_static_routes=dst_static_routes
        )

        return self._rest_api_update(data)

    def delete(self, etht_service_name : str) -> bool:
        return self._rest_api_delete(etht_service_name)