merge from development branch

Author: Michael Hammann

CHANGELOG.md

@@ -0,0 +1,29 @@
+# Changelog of supervisor patch
+
+#### Upcoming Major Changes
+
+#### Upcoming Minor Changes
+
+#### Upcoming Patch Changes
+
+#### 4.2.2-agile5@ar/thirdparty
+- fix: depends_on is not blocking anymore, states are properly recognized, errors are handled
+- feat: Additional process parameter spawn_timeout: Time until the process must be RUNNING if another process depends on it (deafult=60 seconds)
+
+#### 4.2.2-agile4@ar/thirdparty (10/09/2021)
+- depends_on no works correctly when dependency is set to process within a group (group:process)
+
+#### 4.2.2-agile3@ar/thirdparty (08/09/2021)
+- Runningregex, only consideres log output after spawning isntead of the full log file. (http://git.ar.int/dev/sys/sopl/ar-supervisor/-/issues/17)
+- Depends_on now supports to use autostart, uses dicts for the storage of dependencies and programs only start if dependent processes are in RUNNING state(http://git.ar.int/dev/sys/sopl/ar-supervisor/-/issues/15)
+
+
+#### 4.2.2-agile2@ar/thirdparty
+- fix: depends_on flag excepts multiple dependencies for each process now (dev/sys/sopl/ar-supervisor#12)
+- fix: depends_on was only working for first program within a group- now it can be used with all
+- fix: depends_on flag needs now to be used with group:process if process is in group or just process if process is not assigned to any group (dev/sys/sopl/ar-supervisor#11)
+
+
+#### 4.2.2-agile1@ar/thirdparty
+- feat: runningregex: set RUNNING state correctly when process is ready. dev/sys/sopl/ar-supervisor#10
+- feat: depends_on: add optional attribute to only start process, when required processes are started or in RUNNING state. dev/sys/sopl/ar-supervisor#7

supervisor/options.py

@@ -932,7 +932,15 @@ def get(section, opt, *args, **kwargs):
         serverurl = get(section, 'serverurl', None)
         if serverurl and serverurl.strip().upper() == 'AUTO':
             serverurl = None
+        runningregex = get(section, 'runningregex', None)
+        if runningregex:
+            try:
+                runningregex = re.compile(r'.*'+ runningregex)
+            except Exception as e:
+                raise ValueError(
+                    f"program section {section} has invalid runningregex value. Error {e}")
         depends_on = get(section, 'depends_on', None)
+        spawn_timeout = int(get(section, 'spawn_timeout', 60))
 
         # find uid from "user" option
         user = get(section, 'user', None)
@@ -1059,7 +1067,9 @@ def get(section, opt, *args, **kwargs):
                 redirect_stderr=redirect_stderr,
                 environment=environment,
                 serverurl=serverurl,
+                runningregex=runningregex,
                 depends_on=depends_on,
+                spawn_timeout=spawn_timeout,
             )
 
             programs.append(pconfig)
@@ -1878,7 +1888,8 @@ class ProcessConfig(Config):
         'stderr_events_enabled', 'stderr_syslog',
         'stopsignal', 'stopwaitsecs', 'stopasgroup', 'killasgroup',
         'exitcodes', 'redirect_stderr' ]
-    optional_param_names = [ 'environment', 'serverurl', 'depends_on' ]
+    optional_param_names = [ 'environment', 'serverurl', 
+        'runningregex', 'depends_on', 'spawn_timeout' ]
 
     def __init__(self, options, **params):
         self.options = options

supervisor/process.py

@@ -5,6 +5,7 @@
 import shlex
 import time
 import traceback
+import re
 
 from supervisor.compat import maxint
 from supervisor.compat import as_bytes
@@ -22,6 +23,7 @@
 from supervisor.options import decode_wait_status
 from supervisor.options import signame
 from supervisor.options import ProcessException, BadCommand
+from supervisor.options import readFile
 
 from supervisor.dispatchers import EventListenerStates
 
@@ -197,79 +199,97 @@ def record_spawnerr(self, msg):
         self.spawnerr = msg
         self.config.options.logger.info("spawnerr: %s" % msg)
 
-    def spawn(self):
+    def queue_all_dependee_processes(self, supervisor):
+        if self.config.name not in supervisor.process_spawn_set:
+            supervisor.process_spawn_queue.append(self)
+            supervisor.process_spawn_set.add(self.config.name)
+        # all dependees that are not in queue and not in STARTING need to be added to queue.
+        if self.config.depends_on is not None:
+            for dependee in self.config.depends_on.values():
+                if dependee.state is not ProcessStates.RUNNING:
+                    ready_to_be_spawned = False
+                if dependee.state is not ProcessStates.STARTING:
+                    if dependee.config.name not in supervisor.process_spawn_set:
+                        supervisor.process_spawn_queue.append(dependee)
+                        supervisor.process_spawn_set.add(dependee.config.name)
+                        dependee.queue_all_dependee_processes(supervisor)
+
+
+    def spawn(self, supervisor=None):
         """Start the subprocess.  It must not be running already.
 
         Return the process id.  If the fork() call fails, return None.
         """
+        # spawn if all dependees are running - else add to queue if not in queue already
+        ready_to_be_spawned = True
 
-        options = self.config.options
-        processname = as_string(self.config.name)
+            options = self.config.options
+            processname = as_string(self.config.name)
 
-        if self.pid:
-            msg = 'process \'%s\' already running' % processname
-            options.logger.warn(msg)
-            return
+            if self.pid:
+                msg = 'process \'%s\' already running' % processname
+                options.logger.warn(msg)
+                return
 
-        self.killing = False
-        self.spawnerr = None
-        self.exitstatus = None
-        self.system_stop = False
-        self.administrative_stop = False
+            self.killing = False
+            self.spawnerr = None
+            self.exitstatus = None
+            self.system_stop = False
+            self.administrative_stop = False
 
-        self.laststart = time.time()
+            self.laststart = time.time()
 
-        self._assertInState(ProcessStates.EXITED, ProcessStates.FATAL,
-                            ProcessStates.BACKOFF, ProcessStates.STOPPED)
+            self._assertInState(ProcessStates.EXITED, ProcessStates.FATAL,
+                                ProcessStates.BACKOFF, ProcessStates.STOPPED)
 
-        self.change_state(ProcessStates.STARTING)
+            self.change_state(ProcessStates.STARTING)
 
-        try:
-            filename, argv = self.get_execv_args()
-        except ProcessException as what:
-            self.record_spawnerr(what.args[0])
-            self._assertInState(ProcessStates.STARTING)
-            self.change_state(ProcessStates.BACKOFF)
-            return
+            try:
+                filename, argv = self.get_execv_args()
+            except ProcessException as what:
+                self.record_spawnerr(what.args[0])
+                self._assertInState(ProcessStates.STARTING)
+                self.change_state(ProcessStates.BACKOFF)
+                return
 
-        try:
-            self.dispatchers, self.pipes = self.config.make_dispatchers(self)
-        except (OSError, IOError) as why:
-            code = why.args[0]
-            if code == errno.EMFILE:
-                # too many file descriptors open
-                msg = 'too many open files to spawn \'%s\'' % processname
-            else:
-                msg = 'unknown error making dispatchers for \'%s\': %s' % (
-                      processname, errno.errorcode.get(code, code))
-            self.record_spawnerr(msg)
-            self._assertInState(ProcessStates.STARTING)
-            self.change_state(ProcessStates.BACKOFF)
-            return
+            try:
+                self.dispatchers, self.pipes = self.config.make_dispatchers(self)
+            except (OSError, IOError) as why:
+                code = why.args[0]
+                if code == errno.EMFILE:
+                    # too many file descriptors open
+                    msg = 'too many open files to spawn \'%s\'' % processname
+                else:
+                    msg = 'unknown error making dispatchers for \'%s\': %s' % (
+                        processname, errno.errorcode.get(code, code))
+                self.record_spawnerr(msg)
+                self._assertInState(ProcessStates.STARTING)
+                self.change_state(ProcessStates.BACKOFF)
+                return
 
-        try:
-            pid = options.fork()
-        except OSError as why:
-            code = why.args[0]
-            if code == errno.EAGAIN:
-                # process table full
-                msg  = ('Too many processes in process table to spawn \'%s\'' %
-                        processname)
-            else:
-                msg = 'unknown error during fork for \'%s\': %s' % (
-                      processname, errno.errorcode.get(code, code))
-            self.record_spawnerr(msg)
-            self._assertInState(ProcessStates.STARTING)
-            self.change_state(ProcessStates.BACKOFF)
-            options.close_parent_pipes(self.pipes)
-            options.close_child_pipes(self.pipes)
-            return
+            try:
+                pid = options.fork()
+            except OSError as why:
+                code = why.args[0]
+                if code == errno.EAGAIN:
+                    # process table full
+                    msg  = ('Too many processes in process table to spawn \'%s\'' %
+                            processname)
+                else:
+                    msg = 'unknown error during fork for \'%s\': %s' % (
+                        processname, errno.errorcode.get(code, code))
+                self.record_spawnerr(msg)
+                self._assertInState(ProcessStates.STARTING)
+                self.change_state(ProcessStates.BACKOFF)
+                options.close_parent_pipes(self.pipes)
+                options.close_child_pipes(self.pipes)
+                return
 
-        if pid != 0:
-            return self._spawn_as_parent(pid)
+            if pid != 0:
+                return self._spawn_as_parent(pid)
 
-        else:
-            return self._spawn_as_child(filename, argv)
+            else:
+                return self._spawn_as_child(filename, argv)
 
     def _spawn_as_parent(self, pid):
         # Parent
@@ -671,11 +691,11 @@ def transition(self, supervisord_instance=None):
                 if self.config.autorestart:
                     if self.config.autorestart is RestartUnconditionally:
                         # EXITED -> STARTING
-                        self.spawn()
+                        self.spawn(supervisor)
                     else: # autorestart is RestartWhenExitUnexpected
                         if self.exitstatus not in self.config.exitcodes:
                             # EXITED -> STARTING
-                            self.spawn()
+                            self.spawn(supervisor)
             elif state == ProcessStates.STOPPED and not self.laststart:
                 if self.config.autostart:
                     # STOPPED -> STARTING
@@ -691,11 +711,11 @@ def transition(self, supervisord_instance=None):
                 if self.backoff <= self.config.startretries:
                     if now > self.delay:
                         # BACKOFF -> STARTING
-                        self.spawn()
+                        self.spawn(supervisor)
 
         processname = as_string(self.config.name)
         if state == ProcessStates.STARTING:
-            if now - self.laststart > self.config.startsecs:
+            if now - self.laststart > self.config.startsecs and not self.config.runningregex:
                 # STARTING -> RUNNING if the proc has started
                 # successfully and it has stayed up for at least
                 # proc.config.startsecs,
@@ -708,6 +728,27 @@ def transition(self, supervisord_instance=None):
                     '> than %s seconds (startsecs)' % self.config.startsecs)
                 logger.info('success: %s %s' % (processname, msg))
 
+            if self.config.runningregex:
+                logfile = getattr(self.config, 'stdout_logfile')
+                logfile_as_str =  as_string(readFile(logfile, self.log_offset, 0))
+                
+                # delete ascii escape sequence and newlines with regular expression
+                ansi_escape = re.compile(r'(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]|\n')
+                logfile_as_str = ansi_escape.sub('', logfile_as_str)
+
+                # STARTING -> RUNNING if the process has started
+                # successfully and the runningregex is met
+                if self.config.runningregex.match(logfile_as_str):
+                    self.delay = 0
+                    self.backoff = 0
+                    self._assertInState(ProcessStates.STARTING)
+                    self.change_state(ProcessStates.RUNNING)
+                    msg = ('entered RUNNING state, found runningregex in stdout')
+                    logger.info('success: %s %s' % (processname, msg))
+
+
+
+
         if state == ProcessStates.BACKOFF:
             if self.backoff > self.config.startretries:
                 # BACKOFF -> FATAL if the proc has exceeded its number