slapgrid: Process promises with instance python

slapos/grid/promise/runpromises.py

+from __future__ import print_function
+
+import argparse
+import ast
+import os
+import sys
+
+
+# Parse arguments
+
+parser = argparse.ArgumentParser()
+parser.add_argument('--promise-folder', required=True)
+parser.add_argument('--legacy-promise-folder', default=None)
+parser.add_argument('--promise-timeout', type=int, default=20)
+parser.add_argument('--partition-folder', default=None)
+parser.add_argument('--log-folder', default=None)
+parser.add_argument('--force', action='store_true')
+parser.add_argument('--check-anomaly', action='store_true')
+parser.add_argument('--debug', action='store_true')
+parser.add_argument('--master-url', default=None)
+parser.add_argument('--partition-cert', default=None)
+parser.add_argument('--partition-key', default=None)
+parser.add_argument('--partition-id', default=None)
+parser.add_argument('--computer-id', default=None)
+args = parser.parse_args()
+
+
+# Extract slapos.core path and all dependencies from first promise found
+# to import slapos.core
+
+promise_folder = args.promise_folder
+promise_file = next(
+  p for p in os.listdir(promise_folder)
+  if p.endswith('.py') and not p.startswith('__init__')
+)
+with open(os.path.join(promise_folder, promise_file)) as f:
+  promise_content = f.read()
+tree = ast.parse(promise_content, mode='exec')
+
+sys.path[0:0] = eval(compile(ast.Expression(tree.body[1].value), '', 'eval'))
+
+from slapos.grid.promise import PromiseLauncher, PromiseError
+from slapos.cli.entry import SlapOSApp
+
+
+# Configure promise launcher
+# with the same logger as standard slapos command
+
+app = SlapOSApp()
+app.options, _ = app.parser.parse_known_args([])
+app.configure_logging()
+
+config = {k.replace('_', '-') : v for k, v in vars(args).items()}
+promise_checker = PromiseLauncher(config=config, logger=app.log)
+
+
+# Run promises
+# Redirect stdout to stderr (logger only uses stderr already)
+# to reserve stdout for error reporting
+
+out = os.dup(1)
+os.dup2(2, 1)
+
+try:
+  promise_checker.run()
+except Exception as e:
+  os.write(out, str(e))
+  sys.exit(2 if isinstance(e, PromiseError) else 1)

slapos/grid/slapgrid.py

@@ -33,7 +33,6 @@ import pkg_resources
 import random
 import socket
 from io import BytesIO
-import subprocess
 import sys
 import tempfile
 import time
@@ -45,6 +44,11 @@ import shutil
 import six
 import errno
 
+if six.PY3:
+  import subprocess
+else:
+  import subprocess32 as subprocess
+
 if sys.version_info < (2, 6):
   warnings.warn('Used python version (%s) is old and has problems with'
       ' IPv6 connections' % sys.version.split('\n')[0])
@@ -62,7 +66,11 @@ from slapos.grid.svcbackend import (launchSupervisord,
                                     createSupervisordConfiguration,
                                     _getSupervisordConfigurationDirectory,
                                     _getSupervisordSocketPath)
-from slapos.grid.utils import (md5digest, dropPrivileges, SlapPopen, updateFile)
+from slapos.grid.utils import (md5digest,
+                              dropPrivileges,
+                              killProcessTree,
+                              SlapPopen,
+                              updateFile)
 from slapos.grid.promise import PromiseLauncher, PromiseError
 from slapos.grid.promise.generic import PROMISE_LOG_FOLDER_NAME
 from slapos.human import human2bytes
@@ -669,18 +677,18 @@ stderr_logfile_backups=1
     return SLAPGRID_SUCCESS
 
   def _checkPromiseList(self, partition, force=True, check_anomaly=False):
-    instance_path = os.path.join(self.instance_root, partition.partition_id)
+    partition_id = partition.partition_id
+    self.logger.info("Checking %s promises...", partition_id)
+
+    instance_path = os.path.join(self.instance_root, partition_id)
     promise_log_path = os.path.join(instance_path, PROMISE_LOG_FOLDER_NAME)
+    promise_dir = os.path.join(instance_path, 'etc', 'plugin')
+    legacy_promise_dir = os.path.join(instance_path, 'etc', 'promise')
 
-    self.logger.info("Checking %s promises..." % partition.partition_id)
-    uid, gid = None, None
     stat_info = os.stat(instance_path)
-
-    #stat sys call to get statistics informations
     uid = stat_info.st_uid
     gid = stat_info.st_gid
-    promise_dir = os.path.join(instance_path, 'etc', 'plugin')
-    legacy_promise_dir = os.path.join(instance_path, 'etc', 'promise')
+
     promise_config = {
       'promise-folder': promise_dir,
       'legacy-promise-folder': legacy_promise_dir,
@@ -694,12 +702,56 @@ stderr_logfile_backups=1
       'master-url': partition.server_url,
       'partition-cert': partition.cert_file,
       'partition-key': partition.key_file,
-      'partition-id': partition.partition_id,
+      'partition-id': partition_id,
       'computer-id': self.computer_id,
     }
 
-    promise_checker = PromiseLauncher(config=promise_config, logger=self.logger)
-    return promise_checker.run()
+    plugins = sum(
+      1 for p in listifdir(promise_dir)
+      if p.endswith('.py') and not p.startswith('__init__')
+    )
+    instance_python = partition.instance_python
+    if instance_python is not None and plugins:
+      self.logger.info(
+        "Switching to %s's python at %s", partition_id, instance_python)
+      runpromise_script = os.path.join(
+        os.path.dirname(__file__), 'promise', 'runpromises.py')
+      command = [instance_python, runpromise_script]
+      for option, value in promise_config.items():
+        if option in ('uid', 'gid'):
+          continue
+        if isinstance(value, bool):
+          if value:
+            command.append('--' + option)
+        else:
+          command.append('--' + option)
+          command.append(str(value))
+      process = subprocess.Popen(
+        command,
+        preexec_fn=lambda: dropPrivileges(uid, gid, logger=self.logger),
+        cwd=instance_path,
+        stdout=subprocess.PIPE)
+      promises = plugins + len(listifdir(legacy_promise_dir))
+      # Add a timeout margin to let the process kill the promises and cleanup
+      timeout = promises * self.promise_timeout + 10
+      try:
+        # The logger logs everything to stderr, so runpromise redirects
+        # stdout to stderr in case a promise prints to stdout
+        # and reserves stdout to progagate exception messages.
+        out, _ = process.communicate(timeout=timeout)
+        if process.returncode == 2:
+          raise PromiseError(out)
+        elif process.returncode:
+          raise Exception(out)
+        elif out:
+          self.logger.warn('Promise runner unexpected output:\n%s', out)
+      except subprocess.TimeoutExpired:
+        killProcessTree(process.pid, self.logger)
+        # The timeout margin was exceeded but this should be infrequent
+        raise Exception('Promise runner timed out')
+
+    else:
+      return PromiseLauncher(config=promise_config, logger=self.logger).run()
 
   def _endInstallationTransaction(self, computer_partition):
     partition_id = computer_partition.getId()