backend.workers.monitor_workers

 1"""
 2Monitor active workers
 3"""
 4
 5import logging
 6import sys
 7
 8from backend.lib.worker import BasicWorker
 9
10
11class WorkerMonitor(BasicWorker):
12    """
13    Monitor active threads
14
15    This writes a debug log message at an interval listing active 4CAT threads.
16    It can be used to identify workers that do not terminate correctly or to
17    know where a certain worker is stuck if it does not seem to progress. It
18    also shows the 'native ID' of the thread which can be used to find it in
19    e.g. htop or another process monitor to inspect it further.
20
21    If the logging level of 4CAT is not set to DEBUG or lower, this worker does
22    nothing.
23    """
24
25    type = "worker-monitor"
26    max_workers = 1
27
28    @classmethod
29    def ensure_job(cls, config=None):
30        """
31        Run at an interval of 15 seconds.
32        """
33        return {"remote_id": "refresh-items", "interval": 15}
34
35    def work(self):
36        """
37        Monitor active 4CAT threads
38        """
39        if self.log.logger.level > logging.DEBUG:
40            # do nothing if we're not debugging
41            return self.job.finish()
42
43        # try to map active threads to 4CAT workers
44        # and also get the 'native ID' which, at least on linux, is the pid of
45        # the thread - allowing for further inspecting
46        frames = sys._current_frames()
47        thread_id_map = {}
48        for worker_type, workers in self.manager.worker_pool.items():
49            for worker in workers:
50                thread_id_map[worker.ident] = f"{worker_type}/{worker.native_id}"
51
52        monitor_msg = ""
53        for thread_id, frame in frames.items():
54            stack = []
55            while frame:
56                # each frame has a reference to the parent frame (if there is one)
57                # traverse this stack to construct the actual call stack
58                stack.append(
59                    f"{frame.f_code.co_filename.split('/')[-1]}:{frame.f_lineno}:{frame.f_code.co_name}()"
60                )
61                frame = frame.f_back
62
63            stack = " <-- ".join(stack[:-1]) # ignore the very first frame which is never relevant
64            
65            if thread_id in thread_id_map:
66                monitor_msg += f"\n  4CAT worker {thread_id_map[thread_id]} :: {stack}"
67            elif thread_id == self.manager.ident:
68                monitor_msg += f"\n  4CAT main loop :: {stack}"
69            else:
70                monitor_msg += f"\n  4CAT unknown thread {thread_id} :: {stack}"
71
72        self.log.debug(f"Currently {len(frames):,} active 4CAT threads: {monitor_msg}")
73        return self.job.finish()