Merge pull request #106 from cloudblue/support_autoreload_for_development

LITE-25360 support runner autoreload

Author: ffaraone

connect/eaas/runner/constants.py

@@ -134,7 +134,7 @@
     'WebApp': 'Web Application',
 }
 
-PROCESS_CHECK_INTERVAL_SECS = 5
+PROCESS_CHECK_INTERVAL_SECS = 1
 
 LEVEL_TO_FONT_COLOR = {
     'INFO': ('ansi_regular', 'BLUE'),

connect/eaas/runner/main.py

@@ -34,7 +34,12 @@ def start(data):
     if not data.no_validate:
         highest_message_level, tables = validate_extension()
 
-    master = Master(not data.unsecure)
+    master = Master(
+        secure=not data.unsecure,
+        debug=data.debug,
+        no_rich=data.no_rich_logging,
+        reload=data.reload,
+    )
 
     have_features, features = master.get_available_features()
 
@@ -68,11 +73,36 @@ def start(data):
 
 def main():
     parser = argparse.ArgumentParser(prog='cextrun')
-    parser.add_argument('-u', '--unsecure', action='store_true')
-    parser.add_argument('-s', '--split', action='store_true', default=False)
-    parser.add_argument('-d', '--debug', action='store_true', default=False)
-    parser.add_argument('-n', '--no-validate', action='store_true', default=False)
-    parser.add_argument('--no-rich-logging', action='store_true', default=False)
+    parser.add_argument('-u', '--unsecure', action='store_true', help=argparse.SUPPRESS)
+    parser.add_argument('-s', '--split', action='store_true', default=False, help=argparse.SUPPRESS)
+    parser.add_argument(
+        '-d',
+        '--debug',
+        action='store_true',
+        default=False,
+        help='Set the log level to DEBUG.',
+    )
+    parser.add_argument(
+        '-n',
+        '--no-validate',
+        action='store_true',
+        default=False,
+        help='Skip extension validations and start the runner.',
+    )
+    parser.add_argument(
+        '-r',
+        '--reload',
+        action='store_true',
+        default=False,
+        help='Reload the runner when a python file changes (use only for development).',
+    )
+
+    parser.add_argument(
+        '--no-rich-logging',
+        action='store_true',
+        default=False,
+        help='Turn the rich console log to off.',
+    )
     data = parser.parse_args()
     configure_logger(data.debug, data.no_rich_logging)
     if not data.no_validate:

connect/eaas/runner/master.py

@@ -2,7 +2,11 @@
 import threading
 import signal
 import time
-from multiprocessing import Process
+from pathlib import Path
+
+from watchfiles import watch
+from watchfiles.filters import PythonFilter
+from watchfiles.run import start_process
 
 from connect.eaas.runner.config import ConfigHelper
 from connect.eaas.runner.constants import (
@@ -28,6 +32,16 @@
 logger = logging.getLogger('connect.eaas')
 
 
+HANDLED_SIGNALS = (signal.SIGINT, signal.SIGTERM)
+
+
+def _display_path(path):
+    try:
+        return f'"{path.relative_to(Path.cwd())}"'
+    except ValueError:  # pragma: no cover
+        return f'"{path}"'
+
+
 class Master:
 
     HANDLER_CLASSES = {
@@ -44,15 +58,31 @@ class Master:
         ANVILAPP_WORKER: start_anvilapp_worker_process,
     }
 
-    def __init__(self, secure=True):
+    def __init__(self, secure=True, debug=False, no_rich=False, reload=False):
         self.config = ConfigHelper(secure=secure)
         self.handlers = {
             worker_type: self.HANDLER_CLASSES[worker_type](self.config)
             for worker_type in WORKER_TYPES
         }
+        self.reload = reload
+        self.debug = debug
+        self.no_rich = no_rich
         self.workers = {}
-        self.exited_workers = {}
         self.stop_event = threading.Event()
+        self.monitor_event = threading.Event()
+        self.watch_filter = PythonFilter(ignore_paths=None)
+        self.watcher = watch(
+            Path.cwd(),
+            watch_filter=self.watch_filter,
+            stop_event=self.stop_event,
+            yield_on_timeout=True,
+        )
+        self.monitor_thread = None
+        self.setup_signals_handler()
+
+    def setup_signals_handler(self):
+        for sig in HANDLED_SIGNALS:
+            signal.signal(sig, self.handle_signal)
 
     def get_available_features(self):
         have_features = False
@@ -67,38 +97,70 @@ def get_available_features(self):
             features[handler.__class__.__name__] = worker_info
         return have_features, features
 
-    def start_worker_process(self, worker_type, handler):
-        p = Process(
-            target=self.PROCESS_TARGETS[worker_type],
-            args=(handler,),
-        )
-        self.workers[worker_type] = p
-        p.start()
-        logger.info(f'{worker_type.capitalize()} worker pid: {p.pid}')
+    def handle_signal(self, *args, **kwargs):
+        self.stop_event.set()
 
-    def run(self):
+    def start(self):
         for worker_type, handler in self.handlers.items():
             if handler.should_start:
-                self.exited_workers[worker_type] = False
                 self.start_worker_process(worker_type, handler)
+        self.monitor_thread = threading.Thread(target=self.monitor_processes)
+        self.monitor_event.set()
+        self.monitor_thread.start()
 
-        def _terminate(*args, **kwargs):  # pragma: no cover
-            for p in self.workers.values():
-                p.join()
-            self.stop_event.set()
-
-        signal.signal(signal.SIGINT, _terminate)
-        signal.signal(signal.SIGTERM, _terminate)
+    def start_worker_process(self, worker_type, handler):
+        p = start_process(
+            self.PROCESS_TARGETS[worker_type],
+            'function',
+            (handler, self.debug, self.no_rich),
+            {},
+        )
+        self.workers[worker_type] = p
+        logger.info(f'{worker_type.capitalize()} worker pid: {p.pid}')
 
-        while not (self.stop_event.is_set() or all(self.exited_workers.values())):
+    def monitor_processes(self):
+        while self.monitor_event.is_set():
             for worker_type, p in self.workers.items():
                 if not p.is_alive():
                     if p.exitcode != 0:
                         notify_process_restarted(worker_type)
                         logger.info(f'Process of type {worker_type} is dead, restart it')
                         self.start_worker_process(worker_type, self.handlers[worker_type])
                     else:
-                        self.exited_workers[worker_type] = True
                         logger.info(f'{worker_type.capitalize()} worker exited')
 
             time.sleep(PROCESS_CHECK_INTERVAL_SECS)
+
+    def stop(self):
+        self.monitor_event.clear()
+        self.monitor_thread.join()
+        for process in self.workers.values():
+            process.stop(sigint_timeout=5, sigkill_timeout=1)
+            logger.info(f'Consumer process with pid {process.pid} stopped.')
+
+    def restart(self):
+        self.stop()
+        self.start()
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        changes = next(self.watcher)
+        if changes:
+            return list({Path(c[1]) for c in changes})
+        return None
+
+    def run(self):
+        self.start()
+        if self.reload:
+            for files_changed in self:
+                if files_changed:
+                    logger.warning(
+                        'Detected changes in %s. Reloading...',
+                        ', '.join(map(_display_path, files_changed)),
+                    )
+                    self.restart()
+        else:
+            self.stop_event.wait()
+        self.stop()

connect/eaas/runner/workers/anvil.py

@@ -15,7 +15,7 @@
     SetupRequest,
 )
 from connect.eaas.runner.workers.base import WorkerBase
-from connect.eaas.runner.helpers import get_version
+from connect.eaas.runner.helpers import configure_logger, get_version
 
 
 logger = logging.getLogger(__name__)
@@ -66,7 +66,8 @@ async def shutdown(self):
         await super().shutdown()
 
 
-def start_anvilapp_worker_process(handler):
+def start_anvilapp_worker_process(handler, debug, no_rich):
+    configure_logger(debug, no_rich)
     worker = AnvilWorker(handler)
     loop = asyncio.get_event_loop()
     loop.add_signal_handler(

connect/eaas/runner/workers/events.py

@@ -20,7 +20,7 @@
     RESULT_SENDER_MAX_RETRIES,
     RESULT_SENDER_WAIT_GRACE_SECONDS,
 )
-from connect.eaas.runner.helpers import get_version
+from connect.eaas.runner.helpers import configure_logger, get_version
 from connect.eaas.runner.managers import (
     BackgroundTasksManager,
     InteractiveTasksManager,
@@ -201,7 +201,8 @@ def __repr__(self):
         return super().__repr__()
 
 
-def _start_event_worker_process(handler, runner_type):
+def _start_event_worker_process(handler, runner_type, debug, no_rich):
+    configure_logger(debug, no_rich)
     worker = EventsWorker(handler, runner_type=runner_type)
     loop = asyncio.get_event_loop()
     loop.add_signal_handler(
@@ -215,9 +216,9 @@ def _start_event_worker_process(handler, runner_type):
     loop.run_until_complete(worker.start())
 
 
-def start_interactive_worker_process(handler):
-    _start_event_worker_process(handler, 'interactive')
+def start_interactive_worker_process(handler, debug, no_rich):
+    _start_event_worker_process(handler, 'interactive', debug, no_rich)
 
 
-def start_background_worker_process(handler):
-    _start_event_worker_process(handler, 'background')
+def start_background_worker_process(handler, debug, no_rich):
+    _start_event_worker_process(handler, 'background', debug, no_rich)

connect/eaas/runner/workers/web.py

@@ -22,7 +22,7 @@
     WebTask,
 )
 from connect.eaas.runner.workers.base import WorkerBase
-from connect.eaas.runner.helpers import get_version
+from connect.eaas.runner.helpers import configure_logger, get_version
 
 
 logger = logging.getLogger(__name__)
@@ -174,7 +174,8 @@ def build_response(self, task, status, headers, body):
         return message.serialize()
 
 
-def start_webapp_worker_process(handler):
+def start_webapp_worker_process(handler, debug, no_rich):
+    configure_logger(debug, no_rich)
     worker = WebWorker(handler)
     loop = asyncio.get_event_loop()
     loop.add_signal_handler(