Commit 777d7dff authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device component - gNMI OpenConfig Driver: 

- Implemented handlers for MPLS management
- Implemented Network Instance handlers for Connection Point, Endpoint, Vlan
- Added load of MPLS models
parent f2eda4dc

src/device/service/drivers/gnmi_openconfig/handlers/Mpls.py

0 → 100644
+121 −0
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI 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 json, logging, re

from typing import Any, Dict, List, Tuple

from ._Handler import _Handler

from .Tools import get_int, get_str

from .YangHandler import YangHandler



LOGGER = logging.getLogger(__name__)



RE_MPLS_INTERFACE = re.compile(r'^/mpls/interface\[([^\]]+)\]$')

DEFAULT_NETWORK_INSTANCE = 'default'



class MplsHandler(_Handler):

    def get_resource_key(self) -> str: return '/mpls'

    def get_path(self) -> str:

        return '/openconfig-network-instance:network-instances/network-instance/mpls'



    def compose(

        self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False

    ) -> Tuple[str, str]:

        """

        Compose MPLS (global or per-interface) configuration.

        - Global: set LDP router-id (lsr-id) and optional hello timers.

        - Interface: set LDP interface-id and optional hello timers.

        """

        ni_name = get_str(resource_value, 'network_instance', DEFAULT_NETWORK_INSTANCE)

        ni_type = get_str(resource_value, 'network_instance_type')

        if ni_type is None and ni_name == DEFAULT_NETWORK_INSTANCE:

            ni_type = 'openconfig-network-instance-types:DEFAULT_INSTANCE'



        yang_nis : Any = yang_handler.get_data_path('/openconfig-network-instance:network-instances')

        yang_ni : Any = yang_nis.create_path('network-instance[name="{:s}"]'.format(ni_name))

        yang_ni.create_path('config/name', ni_name)

        if ni_type is not None:

            yang_ni.create_path('config/type', ni_type)



        match_if = RE_MPLS_INTERFACE.match(resource_key)

        if delete:

            if match_if:

                if_name = match_if.group(1)

                str_path = (

                    '/network-instances/network-instance[name={:s}]/mpls/signaling-protocols/ldp'

                    '/interface-attributes/interfaces/interface[interface-id={:s}]'

                ).format(ni_name, if_name)

            else:

                str_path = '/network-instances/network-instance[name={:s}]/mpls'.format(ni_name)

            return str_path, json.dumps({})



        if match_if:

            if_name = match_if.group(1)

            hello_interval = get_int(resource_value, 'hello_interval')

            hello_holdtime = get_int(resource_value, 'hello_holdtime')



            path_if_base = (

                'mpls/signaling-protocols/ldp/interface-attributes/interfaces'

                '/interface[interface-id="{:s}"]/config'

            ).format(if_name)

            yang_ni.create_path('{:s}/interface-id'.format(path_if_base), if_name)

            if hello_interval is not None:

                yang_ni.create_path('{:s}/hello-interval'.format(path_if_base), hello_interval)

            if hello_holdtime is not None:

                yang_ni.create_path('{:s}/hello-holdtime'.format(path_if_base), hello_holdtime)



            yang_if : Any = yang_ni.find_path(

                'mpls/signaling-protocols/ldp/interface-attributes/interfaces'

                '/interface[interface-id="{:s}"]'.format(if_name)

            )



            str_path = (

                '/network-instances/network-instance[name={:s}]/mpls/signaling-protocols/ldp'

                '/interface-attributes/interfaces/interface[interface-id={:s}]'

            ).format(ni_name, if_name)

            json_data = json.loads(yang_if.print_mem('json'))

            json_data = json_data['openconfig-network-instance:interface'][0]

            str_data = json.dumps(json_data)

            return str_path, str_data



        # Global LDP configuration

        ldp_cfg = resource_value.get('ldp', resource_value)

        lsr_id = get_str(ldp_cfg, 'lsr_id')

        hello_interval = get_int(ldp_cfg, 'hello_interval')

        hello_holdtime = get_int(ldp_cfg, 'hello_holdtime')



        if lsr_id is not None:

            yang_ni.create_path('mpls/signaling-protocols/ldp/global/config/lsr-id', lsr_id)

        if hello_interval is not None:

            yang_ni.create_path(

                'mpls/signaling-protocols/ldp/interface-attributes/config/hello-interval', hello_interval

            )

        if hello_holdtime is not None:

            yang_ni.create_path(

                'mpls/signaling-protocols/ldp/interface-attributes/config/hello-holdtime', hello_holdtime

            )



        yang_ldp : Any = yang_ni.find_path('mpls/signaling-protocols/ldp')



        str_path = '/network-instances/network-instance[name={:s}]/mpls/signaling-protocols/ldp'.format(ni_name)

        json_data = json.loads(yang_ldp.print_mem('json'))

        json_data = json_data['openconfig-network-instance:ldp']

        str_data = json.dumps(json_data)

        return str_path, str_data



    def parse(

        self, json_data : Dict, yang_handler : YangHandler

    ) -> List[Tuple[str, Dict[str, Any]]]:

        LOGGER.debug('[parse] json_data = %s', json.dumps(json_data))

        # Not required for current tests (L2VPN validation focuses on SetConfig).

        return []

src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceConnectionPoint.py

0 → 100644
+57 −0
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI 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 json, logging

from typing import Any, Dict, List, Tuple

from ._Handler import _Handler

from .Tools import get_str

from .YangHandler import YangHandler



LOGGER = logging.getLogger(__name__)



class NetworkInstanceConnectionPointHandler(_Handler):

    def get_resource_key(self) -> str: return '/network_instance/connection_point'

    def get_path(self) -> str:

        return '/openconfig-network-instance:network-instances/network-instance/connection-points/connection-point'



    def compose(

        self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False

    ) -> Tuple[str, str]:

        ni_name = get_str(resource_value, 'name')

        cp_id   = get_str(resource_value, 'connection_point_id')



        str_path = (

            '/network-instances/network-instance[name={:s}]/connection-points'

            '/connection-point[connection-point-id={:s}]'

        ).format(ni_name, cp_id)

        if delete:

            return str_path, json.dumps({})



        yang_nis : Any = yang_handler.get_data_path('/openconfig-network-instance:network-instances')

        path_cp_base = (

            'network-instance[name="{:s}"]/connection-points'

            '/connection-point[connection-point-id="{:s}"]'

        ).format(ni_name, cp_id)

        yang_nis.create_path('{:s}/config/connection-point-id'.format(path_cp_base), cp_id)



        yang_cp : Any = yang_nis.find_path(path_cp_base)

        json_data = json.loads(yang_cp.print_mem('json'))

        json_data = json_data['openconfig-network-instance:connection-point'][0]

        return str_path, json.dumps(json_data)



    def parse(

        self, json_data : Dict, yang_handler : YangHandler

    ) -> List[Tuple[str, Dict[str, Any]]]:

        LOGGER.debug('[parse] json_data = %s', json.dumps(json_data))

        return []

src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceEndpoint.py

0 → 100644
+89 −0
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI 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 json, logging

from typing import Any, Dict, List, Tuple

from ._Handler import _Handler

from .Tools import get_int, get_str

from .YangHandler import YangHandler



LOGGER = logging.getLogger(__name__)



class NetworkInstanceEndpointHandler(_Handler):

    def get_resource_key(self) -> str: return '/network_instance/connection_point/endpoint'

    def get_path(self) -> str:

        return '/openconfig-network-instance:network-instances/network-instance/connection-points/connection-point/endpoints/endpoint'



    def compose(

        self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False

    ) -> Tuple[str, str]:

        ni_name = get_str(resource_value, 'name')

        cp_id   = get_str(resource_value, 'connection_point_id')

        ep_id   = get_str(resource_value, 'endpoint_id')

        ep_type = get_str(resource_value, 'type')

        precedence = get_int(resource_value, 'precedence')



        str_path = (

            '/network-instances/network-instance[name={:s}]/connection-points/connection-point'

            '[connection-point-id={:s}]/endpoints/endpoint[endpoint-id={:s}]'

        ).format(ni_name, cp_id, ep_id)

        if delete:

            return str_path, json.dumps({})



        if ep_type is not None and ':' not in ep_type:

            ep_type = 'openconfig-network-instance-types:{:s}'.format(ep_type)



        yang_nis : Any = yang_handler.get_data_path('/openconfig-network-instance:network-instances')

        path_ep_base = (

            'network-instance[name="{:s}"]/connection-points/connection-point[connection-point-id="{:s}"]'

            '/endpoints/endpoint[endpoint-id="{:s}"]'

        ).format(ni_name, cp_id, ep_id)

        yang_nis.create_path('{:s}/config/endpoint-id'.format(path_ep_base), ep_id)

        if ep_type is not None:

            yang_nis.create_path('{:s}/config/type'.format(path_ep_base), ep_type)

        if precedence is not None:

            yang_nis.create_path('{:s}/config/precedence'.format(path_ep_base), precedence)



        if ep_type and ep_type.endswith('LOCAL'):

            if_name = get_str(resource_value, 'interface')

            sif_index = get_int(resource_value, 'subinterface', 0)

            if if_name is not None:

                yang_nis.create_path('{:s}/local/config/interface'.format(path_ep_base), if_name)

                yang_nis.create_path('{:s}/local/config/subinterface'.format(path_ep_base), sif_index)

            site_id = get_int(resource_value, 'site_id')

            if site_id is not None:

                yang_nis.create_path('{:s}/local/config/site-id'.format(path_ep_base), site_id)

        elif ep_type and ep_type.endswith('REMOTE'):

            remote_system = get_str(resource_value, 'remote_system')

            vc_id = get_int(resource_value, 'virtual_circuit_id')

            if remote_system is not None:

                yang_nis.create_path('{:s}/remote/config/remote-system'.format(path_ep_base), remote_system)

            if vc_id is not None:

                yang_nis.create_path(

                    '{:s}/remote/config/virtual-circuit-identifier'.format(path_ep_base), vc_id

                )

            site_id = get_int(resource_value, 'site_id')

            if site_id is not None:

                yang_nis.create_path('{:s}/remote/config/site-id'.format(path_ep_base), site_id)



        yang_ep : Any = yang_nis.find_path(path_ep_base)

        json_data = json.loads(yang_ep.print_mem('json'))

        json_data = json_data['openconfig-network-instance:endpoint'][0]

        return str_path, json.dumps(json_data)



    def parse(

        self, json_data : Dict, yang_handler : YangHandler

    ) -> List[Tuple[str, Dict[str, Any]]]:

        LOGGER.debug('[parse] json_data = %s', json.dumps(json_data))

        return []

src/device/service/drivers/gnmi_openconfig/handlers/NetworkInstanceVlan.py

0 → 100644
+66 −0
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI 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 json, logging

from typing import Any, Dict, List, Tuple, Union

from ._Handler import _Handler

from .Tools import get_int, get_str

from .YangHandler import YangHandler



LOGGER = logging.getLogger(__name__)



class NetworkInstanceVlanHandler(_Handler):

    def get_resource_key(self) -> str: return '/network_instance/vlan'

    def get_path(self) -> str:

        return '/openconfig-network-instance:network-instances/network-instance/vlans/vlan'



    def compose(

        self, resource_key : str, resource_value : Dict, yang_handler : YangHandler, delete : bool = False

    ) -> Tuple[str, str]:

        ni_name = get_str(resource_value, 'name', 'default')

        vlan_id = get_int(resource_value, 'vlan_id')

        vlan_name = get_str(resource_value, 'vlan_name')

        str_path = '/network-instances/network-instance[name={:s}]/vlans/vlan[vlan-id={:d}]'.format(

            ni_name, vlan_id

        )

        if delete:

            yang_nis : Any = yang_handler.get_data_path('/openconfig-network-instance:network-instances')

            yang_vlan = yang_nis.find_path('network-instance[name="{:s}"]/vlans/vlan[vlan-id="{:d}"]'.format(

                ni_name, vlan_id))

            if yang_vlan is not None:

                yang_vlan.unlink()

                yang_vlan.free()

            return str_path, json.dumps({})



        yang_nis : Any = yang_handler.get_data_path('/openconfig-network-instance:network-instances')

        yang_ni : Any = yang_nis.create_path('network-instance[name="{:s}"]'.format(ni_name))

        yang_ni.create_path('config/name', ni_name)

        if ni_name == 'default':

            yang_ni.create_path('config/type', 'openconfig-network-instance-types:DEFAULT_INSTANCE')



        yang_vlans : Any = yang_ni.create_path('vlans')

        yang_vlan : Any = yang_vlans.create_path('vlan[vlan-id="{:d}"]'.format(vlan_id))

        yang_vlan.create_path('config/vlan-id', vlan_id)

        if vlan_name is not None:

            yang_vlan.create_path('config/name', vlan_name)



        json_data = json.loads(yang_vlan.print_mem('json'))

        json_data = json_data['openconfig-network-instance:vlan'][0]

        return str_path, json.dumps(json_data)



    def parse(

        self, json_data : Dict, yang_handler : YangHandler

    ) -> List[Tuple[str, Dict[str, Any]]]:

        LOGGER.debug('[parse] json_data = %s', json.dumps(json_data))

        return []

src/device/service/drivers/gnmi_openconfig/handlers/YangHandler.py

+2 −0
Original line number Diff line number Diff line
@@ -42,6 +42,8 @@ YANG_MODULES = [ 
    'openconfig-types',

    'openconfig-policy-types',

    'openconfig-mpls-types',

    'openconfig-mpls',

    'openconfig-mpls-ldp',

    'openconfig-network-instance-types',

    'openconfig-network-instance',

    'openconfig-acl',