Device - gNMI OpenConfig Driver: (eda84a16) · Commits · TFS / controller

src/device/requirements.in

+1 −1

Original line number	Original line	Diff line number	Diff line
	@@ -24,7 +24,7 @@ Flask-HTTPAuth==4.5.0
	Flask-RESTful==0.3.9		Flask-RESTful==0.3.9
	ipaddress		ipaddress
	Jinja2==3.0.3		Jinja2==3.0.3
	libyang==2.8.0		libyang==2.8.4
	macaddress		macaddress
	ncclient==0.6.15		ncclient==0.6.15
	numpy<2.0.0		numpy<2.0.0

src/device/service/drivers/gnmi_openconfig/examples/libyang_examples.py

0 → 100644

+162 −0

Original line number	Original line	Diff line number	Diff line
			# Copyright 2022-2024 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, libyang, logging, os
			from typing import Dict

			logging.basicConfig(level=logging.DEBUG)
			LOGGER = logging.getLogger(__name__)

			YANG_BASE_PATH = '/home/tfs/tfs-ctrl/src/device/service/drivers/gnmi_openconfig/git/openconfig/public'
			YANG_SEARCH_PATHS = ':'.join([
			os.path.join(YANG_BASE_PATH, 'release'),
			os.path.join(YANG_BASE_PATH, 'third_party'),
			])

			YANG_MODULES = [
			'iana-if-type',
			'openconfig-bgp-types',
			'openconfig-vlan-types',

			'openconfig-interfaces',
			'openconfig-if-8021x',
			'openconfig-if-aggregate',
			'openconfig-if-ethernet-ext',
			'openconfig-if-ethernet',
			'openconfig-if-ip-ext',
			'openconfig-if-ip',
			'openconfig-if-poe',
			'openconfig-if-sdn-ext',
			'openconfig-if-tunnel',

			'openconfig-vlan',

			'openconfig-types',
			'openconfig-policy-types',
			'openconfig-mpls-types',
			'openconfig-network-instance-types',
			'openconfig-network-instance',

			'openconfig-platform',
			'openconfig-platform-controller-card',
			'openconfig-platform-cpu',
			'openconfig-platform-ext',
			'openconfig-platform-fabric',
			'openconfig-platform-fan',
			'openconfig-platform-integrated-circuit',
			'openconfig-platform-linecard',
			'openconfig-platform-pipeline-counters',
			'openconfig-platform-port',
			'openconfig-platform-psu',
			'openconfig-platform-software',
			'openconfig-platform-transceiver',
			'openconfig-platform-types',
			]

			class YangHandler:
			def __init__(self) -> None:
			self._yang_context = libyang.Context(YANG_SEARCH_PATHS)
			self._loaded_modules = set()
			for yang_module_name in YANG_MODULES:
			LOGGER.info('Loading module: {:s}'.format(str(yang_module_name)))
			self._yang_context.load_module(yang_module_name).feature_enable_all()
			self._loaded_modules.add(yang_module_name)
			self._data_path_instances = dict()

			def get_data_paths(self) -> Dict[str, libyang.DNode]:
			return self._data_path_instances

			def get_data_path(self, path : str) -> libyang.DNode:
			data_path_instance = self._data_path_instances.get(path)
			if data_path_instance is None:
			data_path_instance = self._yang_context.create_data_path(path)
			self._data_path_instances[path] = data_path_instance
			return data_path_instance

			def destroy(self) -> None:
			self._yang_context.destroy()

			def main():
			yang_handler = YangHandler()

			LOGGER.info('YangHandler Data (before):')
			for path, dnode in yang_handler.get_data_paths().items():
			LOGGER.info('\|-> {:s}: {:s}'.format(str(path), json.dumps(dnode.print_dict())))

			if_name = 'eth1'
			sif_index = 0
			enabled = True
			address_ip = '172.16.0.1'
			address_ip2 = '192.168.0.1'
			address_prefix = 24
			mtu = 1500

			yang_ifs : libyang.DContainer = yang_handler.get_data_path('/openconfig-interfaces:interfaces')
			yang_if_path = 'interface[name="{:s}"]'.format(if_name)
			yang_if : libyang.DContainer = yang_ifs.create_path(yang_if_path)
			yang_if.create_path('config/name', if_name)
			yang_if.create_path('config/enabled', enabled)
			yang_if.create_path('config/mtu', mtu )

			yang_sifs : libyang.DContainer = yang_if.create_path('subinterfaces')
			yang_sif_path = 'subinterface[index="{:d}"]'.format(sif_index)
			yang_sif : libyang.DContainer = yang_sifs.create_path(yang_sif_path)
			yang_sif.create_path('config/index', sif_index)
			yang_sif.create_path('config/enabled', enabled )

			yang_ipv4 : libyang.DContainer = yang_sif.create_path('openconfig-if-ip:ipv4')
			yang_ipv4.create_path('config/enabled', enabled)

			yang_ipv4_addrs : libyang.DContainer = yang_ipv4.create_path('addresses')
			yang_ipv4_addr_path = 'address[ip="{:s}"]'.format(address_ip)
			yang_ipv4_addr : libyang.DContainer = yang_ipv4_addrs.create_path(yang_ipv4_addr_path)
			yang_ipv4_addr.create_path('config/ip', address_ip )
			yang_ipv4_addr.create_path('config/prefix-length', address_prefix)

			yang_ipv4_addr_path2 = 'address[ip="{:s}"]'.format(address_ip2)
			yang_ipv4_addr2 : libyang.DContainer = yang_ipv4_addrs.create_path(yang_ipv4_addr_path2)
			yang_ipv4_addr2.create_path('config/ip', address_ip2 )
			yang_ipv4_addr2.create_path('config/prefix-length', address_prefix)

			str_data = yang_if.print_mem('json')
			json_data = json.loads(str_data)
			json_data = json_data['openconfig-interfaces:interface'][0]
			str_data = json.dumps(json_data, indent=4)
			LOGGER.info('Resulting Request (before unlink): {:s}'.format(str_data))

			yang_ipv4_addr2.unlink()

			root_node : libyang.DContainer = yang_handler.get_data_path('/openconfig-interfaces:interfaces')
			LOGGER.info('root_node={:s}'.format(str(root_node.print_mem('json'))))

			for s in root_node.siblings():
			LOGGER.info('sibling: {:s}'.format(str(s)))

			PATH_TMPL = '/openconfig-interfaces:interfaces/interface[name="{:s}"]/subinterfaces/subinterface[index="{:d}"]'
			yang_sif = root_node.find_path(PATH_TMPL.format(if_name, sif_index))
			if yang_sif is not None:
			LOGGER.info('yang_sif={:s}'.format(str(yang_sif.print_mem('json'))))
			yang_sif.unlink()
			yang_sif.free()

			str_data = yang_if.print_mem('json')
			json_data = json.loads(str_data)
			json_data = json_data['openconfig-interfaces:interface'][0]
			str_data = json.dumps(json_data, indent=4)
			LOGGER.info('Resulting Request (after unlink): {:s}'.format(str_data))

			yang_handler.destroy()

			if __name__ == '__main__':
			main()

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

+18 −40

Original line number	Original line	Diff line number	Diff line
	@@ -12,7 +12,7 @@
	# See the License for the specific language governing permissions and		# See the License for the specific language governing permissions and
	# limitations under the License.		# limitations under the License.

	import json, libyang, logging, queue		import json, libyang, logging
	from typing import Any, Dict, List, Tuple		from typing import Any, Dict, List, Tuple
	from ._Handler import _Handler		from ._Handler import _Handler
	from .Tools import get_bool, get_int, get_str		from .Tools import get_bool, get_int, get_str
	@@ -34,6 +34,22 @@ class InterfaceHandler(_Handler):
	PATH_TMPL = '/interfaces/interface[name={:s}]/subinterfaces/subinterface[index={:d}]'		PATH_TMPL = '/interfaces/interface[name={:s}]/subinterfaces/subinterface[index={:d}]'
	str_path = PATH_TMPL.format(if_name, sif_index)		str_path = PATH_TMPL.format(if_name, sif_index)
	str_data = json.dumps({})		str_data = json.dumps({})

			root_node : libyang.DContainer = yang_handler.get_data_path(
			'/openconfig-interfaces:interfaces'
			)
			yang_sif = root_node.find_path('/'.join([
			'', # add slash at the beginning
			'openconfig-interfaces:interfaces',
			'interface[name="{:s}"]'.format(if_name),
			'subinterfaces',
			'subinterface[index="{:d}"]'.format(sif_index),
			]))
			if yang_sif is not None:
			LOGGER.info('Deleting: {:s}'.format(str(yang_sif.print_mem('json'))))
			yang_sif.unlink()
			yang_sif.free()

	return str_path, str_data		return str_path, str_data

	enabled = get_bool(resource_value, 'enabled', True) # True/False		enabled = get_bool(resource_value, 'enabled', True) # True/False
	@@ -43,81 +59,43 @@ class InterfaceHandler(_Handler):
	address_prefix = get_int (resource_value, 'address_prefix') # 24		address_prefix = get_int (resource_value, 'address_prefix') # 24
	mtu = get_int (resource_value, 'mtu' ) # 1500		mtu = get_int (resource_value, 'mtu' ) # 1500

	objects_to_free = queue.LifoQueue[libyang.DContainer]()

	yang_ifs : libyang.DContainer = yang_handler.get_data_path('/openconfig-interfaces:interfaces')		yang_ifs : libyang.DContainer = yang_handler.get_data_path('/openconfig-interfaces:interfaces')
	objects_to_free.put_nowait(yang_ifs)
	yang_if_path = 'interface[name="{:s}"]'.format(if_name)		yang_if_path = 'interface[name="{:s}"]'.format(if_name)
	yang_if : libyang.DContainer = yang_ifs.create_path(yang_if_path)		yang_if : libyang.DContainer = yang_ifs.create_path(yang_if_path)
	objects_to_free.put_nowait(yang_if)
	yang_if.create_path('config/name', if_name )		yang_if.create_path('config/name', if_name )
	if enabled is not None: yang_if.create_path('config/enabled', enabled)		if enabled is not None: yang_if.create_path('config/enabled', enabled)
	if mtu is not None: yang_if.create_path('config/mtu', mtu)		if mtu is not None: yang_if.create_path('config/mtu', mtu)

	yang_sifs : libyang.DContainer = yang_if.create_path('subinterfaces')		yang_sifs : libyang.DContainer = yang_if.create_path('subinterfaces')
	objects_to_free.put_nowait(yang_sifs)
	yang_sif_path = 'subinterface[index="{:d}"]'.format(sif_index)		yang_sif_path = 'subinterface[index="{:d}"]'.format(sif_index)
	yang_sif : libyang.DContainer = yang_sifs.create_path(yang_sif_path)		yang_sif : libyang.DContainer = yang_sifs.create_path(yang_sif_path)
	objects_to_free.put_nowait(yang_sif)
	yang_sif.create_path('config/index', sif_index)		yang_sif.create_path('config/index', sif_index)
	if enabled is not None: yang_sif.create_path('config/enabled', enabled)		if enabled is not None: yang_sif.create_path('config/enabled', enabled)

	if vlan_id is not None:		if vlan_id is not None:
	yang_subif_vlan : libyang.DContainer = yang_sif.create_path('openconfig-vlan:vlan')		yang_subif_vlan : libyang.DContainer = yang_sif.create_path('openconfig-vlan:vlan')
	objects_to_free.put_nowait(yang_subif_vlan)
	yang_subif_vlan.create_path('match/single-tagged/config/vlan-id', vlan_id)		yang_subif_vlan.create_path('match/single-tagged/config/vlan-id', vlan_id)

	yang_ipv4 : libyang.DContainer = yang_sif.create_path('openconfig-if-ip:ipv4')		yang_ipv4 : libyang.DContainer = yang_sif.create_path('openconfig-if-ip:ipv4')
	objects_to_free.put_nowait(yang_ipv4)
	if enabled is not None: yang_ipv4.create_path('config/enabled', enabled)		if enabled is not None: yang_ipv4.create_path('config/enabled', enabled)

	if address_ip is not None and address_prefix is not None:		if address_ip is not None and address_prefix is not None:
	yang_ipv4_addrs : libyang.DContainer = yang_ipv4.create_path('addresses')		yang_ipv4_addrs : libyang.DContainer = yang_ipv4.create_path('addresses')
	objects_to_free.put_nowait(yang_ipv4_addrs)
	yang_ipv4_addr_path = 'address[ip="{:s}"]'.format(address_ip)		yang_ipv4_addr_path = 'address[ip="{:s}"]'.format(address_ip)
	yang_ipv4_addr : libyang.DContainer = yang_ipv4_addrs.create_path(yang_ipv4_addr_path)		yang_ipv4_addr : libyang.DContainer = yang_ipv4_addrs.create_path(yang_ipv4_addr_path)
	objects_to_free.put_nowait(yang_ipv4_addr)
	yang_ipv4_addr.create_path('config/ip', address_ip)		yang_ipv4_addr.create_path('config/ip', address_ip)
	yang_ipv4_addr.create_path('config/prefix-length', address_prefix)		yang_ipv4_addr.create_path('config/prefix-length', address_prefix)

	LOGGER.info('YangHandler Data:')		LOGGER.info('YangHandler Data:')
	for path, dnode in yang_handler.get_data_paths().items():		for path, dnode in yang_handler.get_data_paths().items():
	LOGGER.debug('\|-> {:s}: {:s}'.format(str(path), json.dumps(dnode.print_dict())))		LOGGER.info('\|-> {:s}: {:s}'.format(str(path), json.dumps(dnode.print_dict())))

	str_path = '/interfaces/interface[name={:s}]'.format(if_name)		str_path = '/interfaces/interface[name={:s}]'.format(if_name)
	str_data = yang_if.print_mem('json')		str_data = yang_if.print_mem('json')
	json_data = json.loads(str_data)		json_data = json.loads(str_data)
	json_data = json_data['openconfig-interfaces:interface'][0]		json_data = json_data['openconfig-interfaces:interface'][0]
	str_data = json.dumps(json_data)		str_data = json.dumps(json_data)

	# List elements to release:
	LOGGER.warning('Objects to release:')
	#LOGGER.warning('Releasing...')
	while not objects_to_free.empty():
	LOGGER.warning('Getting...')
	try:
	obj = objects_to_free.get_nowait()
	if obj is None:
	LOGGER.warning('Item is None')
	continue

	LOGGER.warning('Releasing[type]: {:s}'.format(str(type(obj))))
	LOGGER.warning('Releasing: {:s} => {:s}'.format(
	str(obj.path()), str(obj.print_mem('json'))
	))

	#try:
	# LOGGER.warning('Freeing...')
	# obj.free()
	# LOGGER.warning('Free done')
	#except:
	# LOGGER.exception('Something went wrong...')
	# is_error = True
	# break
	except queue.Empty:
	LOGGER.warning('No more objects...')
	continue
	#LOGGER.warning('Release done')
	return str_path, str_data		return str_path, str_data

	def parse(		def parse(