Merge branch 'feat/367-cttc-end-to-end-integration-test-with-etsi-osm-ro-ietf-l2vpn' into 'develop' (047ace8e) · Commits · TFS / controller

src/common/tests/test_base_event_collector_retry.py

0 → 100644

+72 −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 grpc
		import common.tools.grpc.BaseEventCollector as base_event_collector_module
		from common.proto.context_pb2 import DeviceEvent, EventTypeEnum


		class _FakeRpcError(grpc.RpcError):
		def __init__(self, status_code):
		self._status_code = status_code

		def code(self):
		return self._status_code


		class _FakeStream:
		def __init__(self, events=None):
		self._events = iter(events or [])

		def __iter__(self):
		return self

		def __next__(self):
		return next(self._events)

		def cancel(self):
		pass


		def _create_device_event():
		event = DeviceEvent()
		event.event.event_type = EventTypeEnum.EVENTTYPE_CREATE
		event.event.timestamp.timestamp = 1.0
		event.device_id.device_uuid.uuid = 'dev1'
		return event


		def test_base_event_collector_retries_if_subscription_creation_fails(monkeypatch):
		monkeypatch.setattr(base_event_collector_module.time, 'sleep', lambda _seconds: None)

		state = {'calls': 0}

		def subscription_method(_request):
		state['calls'] += 1
		if state['calls'] == 1:
		raise _FakeRpcError(grpc.StatusCode.UNAVAILABLE)
		if state['calls'] == 2:
		return _FakeStream(events=[_create_device_event()])
		raise _FakeRpcError(grpc.StatusCode.CANCELLED)

		collector = base_event_collector_module.BaseEventCollector()
		collector.install_collector(subscription_method, object())

		collector.start()
		try:
		event = collector.get_event(block=True, timeout=1.0)
		finally:
		collector.stop()

		assert event.device_id.device_uuid.uuid == 'dev1'

src/common/tools/grpc/BaseEventCollector.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -41,8 +41,8 @@ class CollectorThread(threading.Thread):

		def run(self) -> None:
		while not self._terminate.is_set():
		self._stream = self._subscription_func()
		try:
		self._stream = self._subscription_func()
		for event in self._stream:
		if self._log_events_received:
		str_event = grpc_message_to_json_string(event)

src/context/client/EventsCollector.py

+19 −130

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

		import grpc, logging, queue, threading, time
		from typing import Callable
		import logging
		from common.proto.context_pb2 import Empty
		from common.tools.grpc.Tools import grpc_message_to_json_string
		from common.tools.grpc.BaseEventCollector import BaseEventCollector
		from context.client.ContextClient import ContextClient

		LOGGER = logging.getLogger(__name__)
		LOGGER.setLevel(logging.DEBUG)

		class _Collector(threading.Thread):
		def __init__(
		self, subscription_func : Callable, events_queue = queue.PriorityQueue,
		terminate = threading.Event, log_events_received: bool = False
		) -> None:
		super().__init__(daemon=False)
		self._subscription_func = subscription_func
		self._events_queue = events_queue
		self._terminate = terminate
		self._log_events_received = log_events_received
		self._stream = None

		def cancel(self) -> None:
		if self._stream is None: return
		self._stream.cancel()

		def run(self) -> None:
		while not self._terminate.is_set():
		self._stream = self._subscription_func()
		try:
		for event in self._stream:
		if self._log_events_received:
		str_event = grpc_message_to_json_string(event)
		LOGGER.info('[_collect] event: {:s}'.format(str_event))
		timestamp = event.event.timestamp.timestamp
		self._events_queue.put_nowait((timestamp, event))
		except grpc.RpcError as e:
		if e.code() == grpc.StatusCode.UNAVAILABLE:
		LOGGER.info('[_collect] UNAVAILABLE... retrying...')
		time.sleep(0.5)
		continue
		elif e.code() == grpc.StatusCode.CANCELLED:
		break
		else:
		raise # pragma: no cover

		class EventsCollector:
		class EventsCollector(BaseEventCollector):
		def __init__(
		self, context_client : ContextClient,
		log_events_received : bool = False,
		@@ -69,93 +32,19 @@ class EventsCollector:
		activate_slice_collector : bool = True,
		activate_connection_collector : bool = True,
		) -> None:
		self._events_queue = queue.PriorityQueue()
		self._terminate = threading.Event()
		self._log_events_received = log_events_received

		self._context_thread = _Collector(
		lambda: context_client.GetContextEvents(Empty()),
		self._events_queue, self._terminate, self._log_events_received
		) if activate_context_collector else None

		self._topology_thread = _Collector(
		lambda: context_client.GetTopologyEvents(Empty()),
		self._events_queue, self._terminate, self._log_events_received
		) if activate_topology_collector else None

		self._device_thread = _Collector(
		lambda: context_client.GetDeviceEvents(Empty()),
		self._events_queue, self._terminate, self._log_events_received
		) if activate_device_collector else None

		self._link_thread = _Collector(
		lambda: context_client.GetLinkEvents(Empty()),
		self._events_queue, self._terminate, self._log_events_received
		) if activate_link_collector else None

		self._service_thread = _Collector(
		lambda: context_client.GetServiceEvents(Empty()),
		self._events_queue, self._terminate, self._log_events_received
		) if activate_service_collector else None

		self._slice_thread = _Collector(
		lambda: context_client.GetSliceEvents(Empty()),
		self._events_queue, self._terminate, self._log_events_received
		) if activate_slice_collector else None

		self._connection_thread = _Collector(
		lambda: context_client.GetConnectionEvents(Empty()),
		self._events_queue, self._terminate, self._log_events_received
		) if activate_connection_collector else None

		def start(self):
		self._terminate.clear()

		if self._context_thread is not None: self._context_thread.start()
		if self._topology_thread is not None: self._topology_thread.start()
		if self._device_thread is not None: self._device_thread.start()
		if self._link_thread is not None: self._link_thread.start()
		if self._service_thread is not None: self._service_thread.start()
		if self._slice_thread is not None: self._slice_thread.start()
		if self._connection_thread is not None: self._connection_thread.start()

		def get_event(self, block : bool = True, timeout : float = 0.1):
		try:
		_,event = self._events_queue.get(block=block, timeout=timeout)
		return event
		except queue.Empty: # pylint: disable=catching-non-exception
		return None

		def get_events(self, block : bool = True, timeout : float = 0.1, count : int = None):
		events = []
		if count is None:
		while not self._terminate.is_set():
		event = self.get_event(block=block, timeout=timeout)
		if event is None: break
		events.append(event)
		else:
		while len(events) < count:
		if self._terminate.is_set(): break
		event = self.get_event(block=block, timeout=timeout)
		if event is None: continue
		events.append(event)
		return sorted(events, key=lambda e: e.event.timestamp.timestamp)

		def stop(self):
		self._terminate.set()

		if self._context_thread is not None: self._context_thread.cancel()
		if self._topology_thread is not None: self._topology_thread.cancel()
		if self._device_thread is not None: self._device_thread.cancel()
		if self._link_thread is not None: self._link_thread.cancel()
		if self._service_thread is not None: self._service_thread.cancel()
		if self._slice_thread is not None: self._slice_thread.cancel()
		if self._connection_thread is not None: self._connection_thread.cancel()

		if self._context_thread is not None: self._context_thread.join()
		if self._topology_thread is not None: self._topology_thread.join()
		if self._device_thread is not None: self._device_thread.join()
		if self._link_thread is not None: self._link_thread.join()
		if self._service_thread is not None: self._service_thread.join()
		if self._slice_thread is not None: self._slice_thread.join()
		if self._connection_thread is not None: self._connection_thread.join()
		super().__init__()

		if activate_context_collector:
		self.install_collector(context_client.GetContextEvents, Empty(), log_events_received)
		if activate_topology_collector:
		self.install_collector(context_client.GetTopologyEvents, Empty(), log_events_received)
		if activate_device_collector:
		self.install_collector(context_client.GetDeviceEvents, Empty(), log_events_received)
		if activate_link_collector:
		self.install_collector(context_client.GetLinkEvents, Empty(), log_events_received)
		if activate_service_collector:
		self.install_collector(context_client.GetServiceEvents, Empty(), log_events_received)
		if activate_slice_collector:
		self.install_collector(context_client.GetSliceEvents, Empty(), log_events_received)
		if activate_connection_collector:
		self.install_collector(context_client.GetConnectionEvents, Empty(), log_events_received)

src/context/tests/test_events_collector_retry.py

0 → 100644

+84 −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 grpc
		import common.tools.grpc.BaseEventCollector as base_event_collector_module
		import context.client.EventsCollector as context_events_module
		from common.proto.context_pb2 import DeviceEvent, EventTypeEnum


		class _FakeRpcError(grpc.RpcError):
		def __init__(self, status_code):
		self._status_code = status_code

		def code(self):
		return self._status_code


		class _FakeStream:
		def __init__(self, events=None):
		self._events = iter(events or [])

		def __iter__(self):
		return self

		def __next__(self):
		return next(self._events)

		def cancel(self):
		pass


		class _FakeContextClient:
		def __init__(self):
		self.calls = 0

		def GetDeviceEvents(self, _request):
		self.calls += 1
		if self.calls == 1:
		raise _FakeRpcError(grpc.StatusCode.UNAVAILABLE)
		if self.calls == 2:
		return _FakeStream(events=[_create_device_event()])
		raise _FakeRpcError(grpc.StatusCode.CANCELLED)


		def _create_device_event():
		event = DeviceEvent()
		event.event.event_type = EventTypeEnum.EVENTTYPE_CREATE
		event.event.timestamp.timestamp = 1.0
		event.device_id.device_uuid.uuid = 'dev1'
		return event


		def test_events_collector_retries_if_subscription_creation_fails(monkeypatch):
		monkeypatch.setattr(base_event_collector_module.time, 'sleep', lambda _seconds: None)

		collector = context_events_module.EventsCollector(
		_FakeContextClient(),
		activate_context_collector=False,
		activate_topology_collector=False,
		activate_device_collector=True,
		activate_link_collector=False,
		activate_service_collector=False,
		activate_slice_collector=False,
		activate_connection_collector=False,
		)

		collector.start()
		try:
		event = collector.get_event(block=True, timeout=1.0)
		finally:
		collector.stop()

		assert event.device_id.device_uuid.uuid == 'dev1'

src/device/service/drivers/gnmi_openconfig/GnmiOpenConfigDriver.py

+4 −2

Original line number	Diff line number	Diff line
		@@ -17,7 +17,7 @@ from typing import Any, Iterator, List, Optional, Tuple, Union
		from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
		from common.type_checkers.Checkers import chk_type
		from device.service.driver_api._Driver import _Driver
		from .GnmiSessionHandler import GnmiSessionHandler
		from .GnmiSessionHandler import GnmiSessionHandler, INITIAL_TARGET_INFO_RESOURCE_KEY

		DRIVER_NAME = 'gnmi_openconfig'
		METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})
		@@ -51,7 +51,9 @@ class GnmiOpenConfigDriver(_Driver):
		@metered_subclass_method(METRICS_POOL)
		def GetInitialConfig(self) -> List[Tuple[str, Any]]:
		with self.__lock:
		return []
		target_facts = self.__handler.target_facts
		if len(target_facts) == 0: return []
		return [(INITIAL_TARGET_INFO_RESOURCE_KEY, target_facts)]

		@metered_subclass_method(METRICS_POOL)
		def GetConfig(self, resource_keys : List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]: