Initial changes for ABORT signaling feature.

Author: Nathan Lee

pyrunner/core/config.py

@@ -238,6 +238,17 @@ def ctx_file(self):
     else:
       return '{}/{}.ctx'.format(self['temp_dir'], self['app_name'])
 
+  @property
+  def abort_sig_file(self):
+    """
+    Path/filename of job's .sig file.
+    This file should only appear in case any signal is to be sent to a running job.
+    """
+    if not self['temp_dir'] or not self['app_name']:
+      return None
+    else:
+      return '{}/.{}.sig.abort'.format(self['temp_dir'], self['app_name'])
+  
   def source_config_file(self, config_file):
     """
     Sources config file to export environment variables.

pyrunner/core/engine.py

@@ -57,6 +57,12 @@ def initiate(self, **kwargs):
     try:
       while self.register.running_nodes or self.register.pending_nodes:
 
+        # Check for abort signals
+        if os.path.exists(self.config.abort_sig_file):
+          print('ABORT signal received! Terminating all running Workers.')
+          self._abort_all_workers()
+          return -1
+        
         # Poll running nodes for completion/failure
         for node in self.register.running_nodes.copy():
           retcode = node.poll()
@@ -106,9 +112,8 @@ def initiate(self, **kwargs):
     except KeyboardInterrupt:
       print('\nKeyboard Interrupt Received')
       print('\nCancelling Execution')
-      for node in self.register.running_nodes:
-        node.terminate()
-      return
+      self._abort_all_workers()
+      return -1
 
     if not kwargs.get('silent'):
       self._print_final_state(ab_code)
@@ -118,6 +123,13 @@ def initiate(self, **kwargs):
 
     return len(self.register.failed_nodes)
 
+  def _abort_all_workers(self):
+    for node in self.register.running_nodes:
+      node.terminate()
+      #self.register.running_nodes.remove(node)
+      #self.register.aborted_nodes.add(node)
+      #self.register.set_children_defaulted(node)
+  
   def _print_current_state(self):
     elapsed = time.time() - self.start_time
 

pyrunner/core/node.py

@@ -86,10 +86,16 @@ def execute(self):
 
     try:
       worker_class = getattr(importlib.import_module(self.module), self.worker)
+      
+      # Check if provided worker actually extends the Worker class.
       if issubclass(worker_class, Worker):
         worker = worker_class(self.context, self._retcode, self.logfile, self.argv)
+      # If it does not extend the Worker class, initialize a reverse-Worker in which the
+      # worker extends the provided class.
       else:
         worker = self.generate_worker()(self.context, self._retcode, self.logfile, self.argv)
+      
+      # Launch the "run" method of the provided Worker under a new process.
       self._thread = multiprocessing.Process(target=worker.protected_run, daemon=False)
       self._thread.start()
     except Exception as e:
@@ -130,6 +136,10 @@ def poll(self, wait=False):
   def terminate(self):
     if self._thread.is_alive():
       self._thread.terminate()
+      logger = lg.FileLogger(self.logfile)
+      logger.open(False)
+      logger._system_("Keyboard Interrupt (SIGINT) received. Terminating all Worker and exiting.")
+      logger.close()
     return
 
 

pyrunner/core/pyrunner.py

@@ -121,7 +121,7 @@ def plugin_notification(self, obj):
     if not isinstance(obj, notification.Notification): raise Exception('Notification plugin must implement the Notification interface')
     self.notification = obj
 
-  # App lifecycle hooks
+  # App lifecycle hooks/decorators
   def on_create(self, func):
     self._on_create_func = func
   def on_start(self, func):
@@ -192,15 +192,15 @@ def run(self):
 
     emit_notification = True
 
-    # # App lifecycle - SUCCESS
+    # App lifecycle - SUCCESS
     if retcode == 0:
       if self._on_success_func:
         self._on_success_func()
       if not self.config['email_on_success']:
         print('Skipping Email Notification: Property "email_on_success" is set to FALSE.')
         emit_notification = False
-    # # App lifecycle - FAIL
-    else:
+    # App lifecycle - FAIL (<0 is for ABORT or other interrupt)
+    elif retcode > 0:
       if self._on_fail_func:
         self._on_fail_func()
       if not self.config['email_on_fail']:
@@ -265,7 +265,12 @@ def zip_log_files(self, exit_status):
 
     try:
 
-      suffix = 'FAILURE' if exit_status else 'SUCCESS'
+      if exit_status == -1:
+        suffix = 'ABORT'
+      elif exit_status > 0:
+        suffix = 'FAILURE'
+      else:
+        suffix = 'SUCCESS'
 
       zip_file = "{}/{}_{}_{}.zip".format(self.config['log_dir'], self.config['app_name'], constants.EXECUTION_TIMESTAMP, suffix)
       print('Zipping Up Log Files to: {}'.format(zip_file))
@@ -356,9 +361,11 @@ def exec_from(self, id)         : return self.register.exec_from(id)
   def exec_disable(self, id_list) : return self.register.exec_disable(id_list)
 
   def parse_args(self):
+    abort = False
+    
     opt_list = 'c:l:n:e:x:N:D:A:t:drhiv'
     longopt_list = [
-      'setup', 'help', 'nozip', 'interactive',
+      'setup', 'help', 'nozip', 'interactive', 'abort',
       'restart', 'version', 'dryrun', 'debug',
       'preserve-context', 'dump-logs', 'disable-exclusive-jobs',
       'email=', 'email-on-fail=', 'email-on-success=', 'ef=', 'es=',
@@ -421,6 +428,8 @@ def parse_args(self):
         self.disable_exclusive_jobs = True
       elif opt in ['--exec-proc-name']:
         self._init_params['exec_proc_name'] = arg
+      elif opt == '--abort':
+        abort = True
       elif opt in ['--serde']:
         if arg.lower() == 'json':
           self.plugin_serde(serde.JsonSerDe())
@@ -441,6 +450,11 @@ def parse_args(self):
       raise RuntimeError('Config file (app_profile) has not been provided')
     self.config.source_config_file(self._init_params['config_file'])
 
+    if abort:
+      print('Submitting ABORT signal to running job for: {}'.format(self.config['app_name']))
+      open(self.config.abort_sig_file, 'a').close()
+      sys.exit(0)
+    
     # Check if restart is possible (ctllog/ctx files exist)
     if self._init_params['restart'] and not self.is_restartable():
       self._init_params['restart'] = False

pyrunner/logger/abstract.py

@@ -59,6 +59,13 @@ def error(self, text):
     """
     self._emit_('ERROR', text)
 
+  def _system_(self, text):
+    """
+    Write a generic SYSTEM level log message.
+    This is reserved for internal control messages.
+    """
+    self._emit_('SYSTEM', text)
+  
   @abstractmethod
   def restart_message(self, restart_count):
     """

pyrunner/version.py

@@ -1 +1 @@
-__version__ = '4.2.5'
+__version__ = '4.3.0'

setup.py

@@ -30,6 +30,6 @@
   license = 'Apache 2.0',
   long_description = 'Python utility providing text-based workflow manager.',
   entry_points = {
-    'console_scripts': ['pyrunner=pyrunner.cli:main', 'pyrunner-repo=pyrunner.cli:repo']
+    'console_scripts': ['pyrunner=pyrunner.cli:main']
   }
 )