librecipe: new 'sig_ign' & 'redirect' parameters to createWrapper

- The 'wrapper' recipe is also extended to support them.
- The 'neoppod' recipe uses 'sig_ign' to avoid a potential kill by
  logrotate at startup.

See merge request !1724

slapos/recipe/librecipe/execute.py

 from __future__ import print_function
 
+import atexit
 import errno
 import sys
 import os
+import select
 import signal
 import subprocess
 import time
@@ -61,12 +63,43 @@ def _libc():
       raise OSError(e, os.strerror(e))
   return mount, unshare
 
+def parse_signal(sig):
+  try:
+    try:
+      a, b = sig.split('+', 1)
+    except ValueError:
+      a = sig
+      b = 0
+    else:
+      b = int(b)
+    if a[0] != '_':
+      return getattr(signal, 'SIG' + a) + b
+  except Exception:
+    raise ValueError("invalid signal value: %s" % sig)
+
 def generic_exec(args, extra_environ=None, wait_list=None,
                  pidfile=None, reserve_cpu=False, private_tmpfs=(),
-                 #shebang_workaround=False, # XXX: still needed ?
+                 # signal name (e.g. "RTMIN+1") that should be specified
+                 # if pidfile (defaults to "USR1" if redirect)
+                 sig_ign=None,
+                 # redirect stdout and/or stderr to files; upon USR1,
+                 # these files are reopened (suitable for log rotation)
+                 # and an optional signal is sent to the spawned process
+                 redirect=None, # (signal, stdout, stderr)
                  ):
+  """
+  All the SIG_IGN stuff is to avoid being killed between:
+  - the moment some external software (like a logrotate configuration
+    snippet) can know the PID via pidfile;
+  - and when the process being starting sets up its signal handler.
+  """
   args = list(args)
 
+  if redirect and not sig_ign:
+    sig_ign = "USR1"
+  if sig_ign:
+    sig_ign = parse_signal(sig_ign)
+    signal.signal(sig_ign, signal.SIG_IGN)
   if pidfile:
     import psutil
     try:
@@ -80,7 +113,14 @@ def generic_exec(args, extra_environ=None, wait_list=None,
       n = len(args)
       for i in six.moves.xrange(1+len(running)-n):
         if args == running[i:n+i]:
-          sys.exit("Already running with pid %s." % pid)
+          return "Already running with pid %s." % pid
+    @atexit.register # some best effort clean-up, and it is
+    def _():         # expected that it does nothing upon execv.
+      try:
+        os.unlink(pidfile)
+      except OSError as e:
+        if e.errno != errno.ENOENT:
+          raise
     with open(pidfile, 'w') as f:
       f.write(str(os.getpid()))
 
@@ -95,27 +135,99 @@ def generic_exec(args, extra_environ=None, wait_list=None,
   if wait_list:
     _wait_files_creation(wait_list)
 
-  if private_tmpfs:
-    mount, unshare = _libc()
-    CLONE_NEWNS   = 0x00020000
-    CLONE_NEWUSER = 0x10000000
-    uid = os.getuid()
-    gid = os.getgid()
-    unshare(CLONE_NEWUSER |CLONE_NEWNS)
-    with open('/proc/self/setgroups', 'w') as f:
-      f.write('deny')
-    with open('/proc/self/uid_map', 'w') as f:
-      f.write('%s %s 1' % (uid, uid))
-    with open('/proc/self/gid_map', 'w') as f:
-      f.write('%s %s 1' % (gid, gid))
-    for size, path in private_tmpfs:
-      try:
-        os.mkdir(path)
-      except OSError as e:
-        if e.errno != errno.EEXIST:
+  def preexec_fn():
+    if private_tmpfs:
+      mount, unshare = _libc()
+      CLONE_NEWNS   = 0x00020000
+      CLONE_NEWUSER = 0x10000000
+      uid = os.getuid()
+      gid = os.getgid()
+      unshare(CLONE_NEWUSER |CLONE_NEWNS)
+      with open('/proc/self/setgroups', 'w') as f:
+        f.write('deny')
+      with open('/proc/self/uid_map', 'w') as f:
+        f.write('%s %s 1' % (uid, uid))
+      with open('/proc/self/gid_map', 'w') as f:
+        f.write('%s %s 1' % (gid, gid))
+      for size, path in private_tmpfs:
+        try:
+          os.mkdir(path)
+        except OSError as e:
+          if e.errno != errno.EEXIST:
+            raise
+        mount(b'tmpfs', path.encode(), b'tmpfs', 0, ('size=' + size).encode())
+    if redirect:
+      if sig != sig_ign:
+        signal.signal(sig_ign, signal.SIG_DFL)
+        signal.signal(sig, signal.SIG_IGN)
+      for fds in dup2:
+        os.dup2(*fds)
+
+  if redirect:
+    if extra_environ:
+      env = os.environ.copy()
+      env.update(extra_environ)
+    else:
+      env = None
+    sig, stdout, stderr = redirect
+    sig = parse_signal(sig)
+    r, trigger = os.pipe()
+    rfds = [r]
+    logs = []
+    dup2 = []
+    def reopen():
+      new = []
+      for path, fd in logs:
+        os.close(fd)
+        new.append((path,
+            os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o666)))
+      logs[:] = new
+    def setup(fd, path):
+      r, w = os.pipe()
+      if w != fd:
+        dup2.append((w, fd))
+      logs.append((path, w))
+      rfds.append(r)
+    if stdout:
+      setup(1, stdout)
+    if stderr:
+      if stderr == stdout:
+        dup2.append((1, 2))
+      else:
+        setup(2, stderr)
+    # First, preexec_fn is called, then close_fds is processed,
+    # and at last, Popen returns.
+    process = subprocess.Popen(args, preexec_fn=preexec_fn, env=env,
+                               close_fds=True) # PY3: this is the default
+    def sighandler(*_):
+      if sig:
+        process.send_signal(sig)
+      os.write(trigger, b'\0')
+    signal.signal(sig_ign, sighandler)
+    reopen()
+    while True:
+      try: # PY3: select internally retries on EINTR
+        r = select.select(rfds, (), ())[0]
+      except select.error as e:
+        if e.args[0] != errno.EINTR:
           raise
-      mount(b'tmpfs', path.encode(), b'tmpfs', 0, ('size=' + size).encode())
+        assert six.PY2
+        continue
+      for r in r:
+        d = os.read(r, 1024)
+        i = rfds.index(r) - 1
+        if i < 0:
+          reopen()
+        elif d:
+          os.write(logs[i][1], d)
+        else:
+          os.close(logs.pop(i)[1])
+          os.close(rfds.pop(i+1))
+          if not logs:
+            signal.signal(sig_ign, signal.SIG_IGN)
+            return process.wait()
 
+  preexec_fn()
   if extra_environ:
     env = os.environ.copy()
     env.update(extra_environ)

slapos/recipe/librecipe/generic.py

@@ -153,10 +153,12 @@ class GenericBaseRecipe(object):
         private_tmpfs.append(tuple(x))
     return private_tmpfs
 
-  def createWrapper(self, path, args, env=None, **kw):
+  def createWrapper(self, path, args, env=None, sig_ign=None, **kw):
     """Create a wrapper script for process replacement"""
     assert args
     if kw:
+      if sig_ign:
+        kw['sig_ign'] = sig_ign
       return self.createPythonScript(path,
         'slapos.recipe.librecipe.execute.generic_exec',
         (args, env) if env else (args,), kw)
@@ -167,8 +169,10 @@ class GenericBaseRecipe(object):
     # here (note that this can't be done correctly with a POSIX shell, because
     # the process can't be given a name).
 
-    lines = ['#!/bin/sh']
+    lines = ['#!/bin/sh -e']
 
+    if sig_ign:
+      lines.append("trap '' " + sig_ign)
     if env:
       for k, v in sorted(six.iteritems(env)):
         lines.append('export %s=%s' % (k, shlex.quote(v)))

slapos/recipe/neoppod.py

@@ -98,8 +98,9 @@ class NeoBaseRecipe(GenericBaseRecipe):
         environment[k.rstrip()] = v.lstrip()
 
     private_tmpfs = self.parsePrivateTmpfs()
-    kw = {'private_tmpfs': private_tmpfs} if private_tmpfs else {}
-    return self.createWrapper(options['wrapper'], args, env=environment, **kw)
+    return self.createWrapper(
+      options['wrapper'], args, env=environment, sig_ign="RTMIN+1",
+      **{'private_tmpfs': private_tmpfs} if private_tmpfs else {})
 
   def _getBindingAddress(self):
     options = self.options

slapos/recipe/wrapper.py

@@ -27,6 +27,7 @@
 
 import os, shlex
 from slapos.recipe.librecipe import GenericBaseRecipe, generateHashFromFiles
+from slapos.recipe.librecipe.execute import parse_signal
 from zc.buildout import UserError
 
 class Recipe(GenericBaseRecipe):
@@ -40,6 +41,9 @@ class Recipe(GenericBaseRecipe):
     :param str pidfile: path to pidfile ensure exclusivity for the process
     :param lines private-tmpfs: list of "<size> <path>" private tmpfs, using user namespaces
     :param bool reserve-cpu: command will ask for an exclusive CPU core
+    :param str sig-ign: see slapos.recipe.librecipe.execute.generic_exec
+    :param str redirect-{signal,stdout,stderr}: set 'redirect' param
+        to slapos.recipe.librecipe.execute.generic_exec
     """
 
     _existing = ()
@@ -60,6 +64,20 @@ class Recipe(GenericBaseRecipe):
           hash_files = hash_files.split()
           options['__hash_files__'] = generateHashFromFiles(hash_files)
           self.hash_files += hash_files
+        sig = options.get('sig-ign')
+        if sig:
+          parse_signal(sig)
+          self.sig_ign = sig
+        redirect = tuple(options.get('redirect-' + x)
+                         for x in ('signal', 'stdout', 'stderr',))
+        if any(redirect):
+          sig = redirect[0]
+          if sig:
+            parse_signal(sig)
+            if not any(redirect[1:]):
+              raise UserError(
+                "redirect-signal without redirecting any output makes no sense")
+          self.redirect = redirect
 
     def getWrapperPath(self):
         wrapper_path = self.options['wrapper-path']
@@ -94,6 +112,11 @@ class Recipe(GenericBaseRecipe):
           kw['private_tmpfs'] = private_tmpfs
         if self.isTrueValue(options.get('reserve-cpu')):
           kw['reserve_cpu'] = True
+        for x in 'redirect', 'sig_ign':
+          try:
+            kw[x] = getattr(self, x)
+          except AttributeError:
+            pass
         return self.createWrapper(self.getWrapperPath(),
                                   args, environment, **kw)