feat(telemetry): ECOC2026-Enhance subscription management and scheduler... (a420c390) · Commits · TFS / controller

src/telemetry/backend/service/TelemetryBackendService.py

+15 −5

Original line number	Diff line number	Diff line
		@@ -58,6 +58,7 @@ class TelemetryBackendService(GenericGrpcService):
		self.kpi_manager_client = KpiManagerClient()
		self.active_jobs = {}
		self.active_collectors = {}
		self.active_subscriptions = {}

		def install_servicers(self):
		threading.Thread(target=self.RequestListener).start()
		@@ -197,13 +198,17 @@ class TelemetryBackendService(GenericGrpcService):
		raise status
		else:
		LOGGER.info(f"Subscription successful for KPI ID: {kpi_id} - Status: {status}")
		self.active_subscriptions[collector_id] = [
		subscription[0] for subscription in resource_to_subscribe
		]

		for (timestamp, resource_key, value) in device_collector.GetState(duration=duration, blocking=True):
		LOGGER.info(f"KPI ID: {kpi_id} - resource={resource_key} value={value}")
		self.GenerateKpiValue(collector_id, kpi_id, value)
		for (timestamp, resource_key, value) in device_collector.GetState(
		duration=duration, blocking=True, terminate=stop_event
		):
		if stop_event.is_set():
		device_collector.Disconnect()
		break
		LOGGER.info(f"KPI ID: {kpi_id} - resource={resource_key} value={value}")
		self.GenerateKpiValue(collector_id, kpi_id, value)

		def GenerateKpiValue(self, collector_id: str, kpi_id: str, measured_kpi_value: Any):
		"""
		@@ -240,8 +245,13 @@ class TelemetryBackendService(GenericGrpcService):
		stop_event.set()
		LOGGER.info(f"Job {job_id} terminated.")
		device_collector = self.active_collectors.pop(job_id, None)
		subscription_ids = self.active_subscriptions.pop(job_id, [])
		if device_collector is not None:
		if device_collector.UnsubscribeState(job_id):
		unsubscribe_status = [
		device_collector.UnsubscribeState(subscription_id)
		for subscription_id in (subscription_ids or [job_id])
		]
		if unsubscribe_status and all(unsubscribe_status):
		LOGGER.info(f"Unsubscribed from collector: {job_id}")
		else:
		LOGGER.warning(f"Failed to unsubscribe from collector: {job_id}")

src/telemetry/backend/service/collectors/netconf_oc/NetconfOCCollector.py

+35 −6

Original line number	Diff line number	Diff line
		@@ -61,12 +61,41 @@ class NetconfOpenConfigCollector(_Collector):
		self._session : Optional[Manager] = None
		self._session_lock : threading.Lock = threading.Lock()
		self._out_samples : queue.Queue = queue.Queue()
		self._scheduler = BackgroundScheduler(daemon=True)
		self._scheduler.configure(
		self._scheduler = self._build_scheduler()

		def _build_scheduler(self) -> BackgroundScheduler:
		scheduler = BackgroundScheduler(daemon=True)
		scheduler.configure(
		jobstores = {'default': MemoryJobStore()},
		executors = {'default': ThreadPoolExecutor(max_workers=1)},
		timezone = pytz.utc,
		)
		return scheduler

		def _scheduler_executor_shutdown(self) -> bool:
		try:
		executor = self._scheduler._executors.get('default') # pylint: disable=protected-access
		executor_pool = getattr(executor, '_pool', None)
		return bool(getattr(executor_pool, '_shutdown', False))
		except Exception:
		return False

		def _ensure_scheduler_started(self) -> None:
		if self._scheduler_executor_shutdown():
		LOGGER.info(
		"Recreating NETCONF OpenConfig scheduler for %s:%s after executor shutdown",
		self._address,
		self._port,
		)
		try:
		if self._scheduler.running:
		self._scheduler.shutdown(wait=False)
		except Exception:
		LOGGER.debug("Ignoring error while replacing stopped NETCONF scheduler", exc_info=True)
		self._scheduler = self._build_scheduler()

		if not self._scheduler.running:
		self._scheduler.start()

		# ------------------------------------------------------------------
		# Connection lifecycle
		@@ -75,8 +104,7 @@ class NetconfOpenConfigCollector(_Collector):
		def Connect(self) -> bool:
		"""Open a NETCONF-over-SSH session and start the background scheduler."""
		if self._session is not None and getattr(self._session, 'connected', True):
		if not self._scheduler.running:
		self._scheduler.start()
		self._ensure_scheduler_started()
		LOGGER.info(
		"NetconfOpenConfigCollector already connected to %s:%s",
		self._address,
		@@ -96,8 +124,7 @@ class NetconfOpenConfigCollector(_Collector):
		allow_agent = False,
		look_for_keys = False,
		)
		if not self._scheduler.running:
		self._scheduler.start()
		self._ensure_scheduler_started()
		LOGGER.info("NetconfOpenConfigCollector connected to %s:%s", self._address, self._port)
		return True
		except Exception:
		@@ -127,6 +154,7 @@ class NetconfOpenConfigCollector(_Collector):
		)
		finally:
		self._session = None
		self._scheduler = self._build_scheduler()
		return True

		# ------------------------------------------------------------------
		@@ -172,6 +200,7 @@ class NetconfOpenConfigCollector(_Collector):
		results: List[Union[bool, Exception]] = []
		for sub_id, endpoint_dict, duration_s, interval_s in subscriptions:
		try:
		self._ensure_scheduler_started()
		channel_name = endpoint_dict['endpoint']
		kpi_code = endpoint_dict['kpi']
		end_date = datetime.now(pytz.utc) + timedelta(seconds=float(duration_s))