def _monitor_cpu()

in src/dubbo/cluster/monitor/cpu.py [0:0]

• 22 lines of code
• 13 McCabe index (conditional complexity)

    def _monitor_cpu(self) -> None:
        """
        Monitor the CPU usage.
        """
        while True:
            # get available invokers
            available_invokers = {url: invoker for url, invoker in self._invokers.items() if invoker.is_available()}

            # update the running invokers
            self._running_invokers = available_invokers

            # update the CPU usage
            with self._usages_lock:
                self._cpu_usages = {
                    invoker: usage
                    for invoker, usage in self._cpu_usages.items()
                    if invoker in available_invokers.values()
                }

            # get the CPU usage for each invoker
            for url, invoker in self._running_invokers.items():
                if invoker.is_available():
                    try:
                        result = invoker.invoke(_cpu_invocation)
                        cpu_usage = float(result.value().decode("utf-8"))
                        self._cpu_usages[invoker] = cpu_usage
                    except Exception:
                        _LOGGER.error("Failed to get the CPU usage for invoker %s: %s", url, str)
                        # remove the cpu usage
                        self._remove_cpu_usage(invoker)

            # wait for the interval or stop
            if self._stop_event.wait(self._interval):
                _LOGGER.info("The CPU monitor has been stopped.")
                break