WIP: Move manager out from slapformat and add security

slapos/README.manager.txt

+slapos.manager
+==============
+
+Manager is a plugin-like class that is being run in multiple phases of slapos node lifecycle.
+
+-  **format**, manager can format additionally the underlaying OS 
+-  **software**, manager can react on software installation
+-  **instance**, manager can update instance runtime frequently
+
+Constructor will receive configuration of current stage. Then each method receives
+object most related to the current operation. For details see <slapos/manager/interface.py>.
+

slapos/grid/slapgrid.py

@@ -49,6 +49,7 @@ if sys.version_info < (2, 6):
 
 from lxml import etree
 
+from slapos import manager as slapmanager
 from slapos.slap.slap import NotFoundError
 from slapos.slap.slap import ServerError
 from slapos.slap.slap import COMPUTER_PARTITION_REQUEST_LIST_TEMPLATE_FILENAME
@@ -274,7 +275,8 @@ def create_slapgrid_object(options, logger):
                   instance_min_free_space=instance_min_free_space,
                   instance_storage_home=op.get('instance_storage_home'),
                   ipv4_global_network=op.get('ipv4_global_network'),
-                  firewall_conf=op.get('firewall'))
+                  firewall_conf=op.get('firewall'),
+                  config=options)
 
 
 def check_required_only_partitions(existing, required):
@@ -333,6 +335,7 @@ class Slapgrid(object):
                instance_storage_home=None,
                ipv4_global_network=None,
                firewall_conf={},
+               config=None,
                ):
     """Makes easy initialisation of class parameters"""
     # Parses arguments
@@ -395,7 +398,8 @@ class Slapgrid(object):
     else:
       self.ipv4_global_network= ""
     self.firewall_conf = firewall_conf
-
+    self.config = config
+    self._manager_list = slapmanager.from_config(config)
 
   def _getWatchdogLine(self):
     invocation_list = [WATCHDOG_PATH]
@@ -553,6 +557,11 @@ stderr_logfile_backups=1
             shadir_cert_file=self.shadir_cert_file,
             shadir_key_file=self.shadir_key_file,
             software_min_free_space=self.software_min_free_space)
+
+        # call manager for every software release
+        for manager in self._manager_list:
+          manager.software(software)
+
         if state == 'available':
           completed_tag = os.path.join(software_path, '.completed')
           if (self.develop or (not os.path.exists(completed_tag) and
@@ -1075,6 +1084,10 @@ stderr_logfile_backups=1
         partition_ip_list = parameter_dict['ip_list'] + parameter_dict.get(
                                                             'full_ip_list', [])
 
+      # call manager for every software release
+      for manager in self._manager_list:
+        manager.instance(partition)
+
       if computer_partition_state == COMPUTER_PARTITION_STARTED_STATE:
         local_partition.install()
         computer_partition.available()

slapos/manager/__init__.py

+# coding: utf-8
+import re
+import importlib 
+
+from zope.interface import declarations
+
+config_option = "manager_list"
+
+
+def load_manager(name):
+    """Load a manager from local files if it exists."""
+    if re.match(r'[a-zA-Z_]', name) is None:
+      raise ValueError("Manager name \"{!s}\" is not allowed! Must contain only letters and \"_\"".
+        format(name))
+
+    from slapos.manager import interface
+    manager_module = importlib.import_module("slapos.manager." + name)
+    if not hasattr(manager_module, "Manager"):
+        raise AttributeError("Manager class in {} has to be called \"Manager\"".format(
+            name))
+    if not interface.IManager.implementedBy(manager_module.Manager):
+        raise RuntimeError("Manager class in {} must zope.interface.implements \"IManager\"".format(
+            name))
+
+    return manager_module.Manager
+
+
+def from_config(config):
+    """Return list of instances of managers allowed from the config."""
+    name_list = config.get(config_option, "").split()
+    return [load_manager(name)(config) for name in name_list]
\ No newline at end of file

slapos/manager/cpuset.py

+# coding: utf-8
+import logging
+import os
+import os.path
+
+from zope import interface as zope_interface
+from slapos.manager import interface
+
+logger = logging.getLogger(__name__)
+
+
+class Manager(object):
+  """Manage cgroup's cpuset in terms on initializing and runtime operations.
+
+  CPUSET manager moves PIDs between CPU cores using Linux cgroup system.
+
+  In order to use this feature put "cpuset" into "manager_list" into your slapos
+  configuration file inside [slapos] section.
+
+  TODO: there is no limit on number of reserved cores per user.
+  """
+  zope_interface.implements(interface.IManager)
+
+  cpu_exclusive_file = ".slapos-cpu-exclusive"
+  cpuset_path = "/sys/fs/cgroup/cpuset/"
+  task_write_mode = "wt"
+  config_power_user_option = "power_user_list"
+
+  def __init__(self, config):
+    """Retain access to dict-like configuration."""
+    self.config = config
+
+  def software(self, software):
+    """We don't need to mingle with software."""
+    pass
+
+  def format(self, computer):
+    """Create cgroup folder per-CPU with exclusive access to the CPU.
+
+    - Those folders are "/sys/fs/cgroup/cpuset/cpu<N>".
+    """
+    if not os.path.exists("/sys/fs/cgroup/cpuset/cpuset.cpus"):
+      logger.warning("CPUSet Manager cannot format computer because cgroups do not exist.")
+      return 
+
+    for cpu in self._cpu_id_list():
+      cpu_path = self._prepare_folder(
+        os.path.join(self.cpuset_path, "cpu" + str(cpu)))
+      with open(cpu_path + "/cpuset.cpus", "wt") as fx:
+        fx.write(str(cpu))  # this cgroup manages only this cpu
+      with open(cpu_path + "/cpuset.cpu_exclusive", "wt") as fx:
+        fx.write("1")  # manages it exclusively
+      with open(cpu_path + "/cpuset.mems", "wt") as fx:
+        fx.write("0")  # it doesn't work without that
+
+  def instance(self, partition):
+    """Control runtime state of the computer."""
+
+    if not os.path.exists(os.path.join(self.cpuset_path, "cpu0")):
+      # check whether the computer was formatted
+      logger.warning("CGROUP's CPUSET Manager cannot update computer because it is not cpuset-formatted.")
+      return
+
+    request_file = os.path.join(partition.instance_path, self.cpu_exclusive_file)
+    if not os.path.exists(request_file) or not read_file(request_file):
+      # This instance does not ask for cpu exclusive access
+      return
+
+    # Gather list of users allowed to request exlusive cores
+    power_user_list = self.config.get(self.config_power_user_option, "").split()
+    uid, gid = partition.getUserGroupId()
+    uname = pwd.getpwuid(uid).pw_name
+    if uname not in power_user_list:
+      logger.warning("User {} not allowed to modify cpuset! "
+                     "Allowed users are in {} option in config file.".format(
+                        uname, self.config_power_user_option))
+      return
+
+    # prepare paths to tasks file for all and per-cpu
+    tasks_file = os.path.join(self.cpuset_path, "tasks")
+    cpu_tasks_file_list = [os.path.join(cpu_folder, "tasks")
+                           for cpu_folder in self._cpu_folder_list()]
+
+    # Gather exclusive CPU usage map {username: set[cpu_id]}
+    cpu_usage = defaultdict(set)
+    for cpu_id in self._cpu_id_list()[1:]:  # skip the first public CPU
+      pids = [int(pid)
+              for pid in read_file(cpu_tasks_file_list[cpu_id]).splitlines()]
+      for pid in pids:
+        process = psutil.Process(pid)
+        cpu_usage[process.username()].add(cpu_id)
+
+    # Move all PIDs from the pool of all CPUs onto the first exclusive CPU.
+    running_list = sorted(list(map(int, read_file(tasks_file).split())), reverse=True)
+    first_cpu = self._cpu_id_list()[0]
+    success_set, refused_set = set(), set()
+    for pid in running_list:
+      try:
+        self._move_task(pid, first_cpu)
+        success_set.add(pid)
+        time.sleep(0.01)
+      except IOError as e:
+        refused_set.add(pid)
+    logger.debug("Refused to move {:d} PIDs: {!s}\n"
+                 "Suceeded in moving {:d} PIDs {!s}\n".format(
+                     len(refused_set), refused_set, len(success_set), success_set))
+
+    cpu_list = self._cpu_folder_list()
+    generic_cpu_path = cpu_folder_list[0]
+    exclusive_cpu_path_list = cpu_folder_list[1:]
+
+    # Gather all running PIDs for filtering out stale PIDs
+    running_pid_set = set(running_list)
+    running_pid_set.update(map(int, read_file(cpu_tasks_file_list[0]).split()))
+
+    # gather already exclusively running PIDs
+    exclusive_pid_set = set()
+    for cpu_tasks_file in cpu_tasks_file_list[1:]:
+      exclusive_pid_set.update(map(int, read_content(cpu_tasks_file).split()))
+
+    # Move processes to their demanded exclusive CPUs
+    with open(request_file, "rt") as fi:
+      # take such PIDs which are either really running or are not already exclusive
+      request_pid_list = [int(pid) for pid in fi.read().split()
+                          if int(pid) in running_pid_set or int(pid) not in exclusive_pid_set]
+    with open(request_file, "wt") as fo:
+      fo.write("")  # empty file (we will write back only PIDs which weren't moved)
+    for request_pid in request_pid_list:
+      assigned_cpu = self._move_to_exclusive_cpu(request_pid)
+      if assigned_cpu < 0:
+        # if no exclusive CPU was assigned - write the PID back and try other time
+        with open(request_file, "at") as fo:
+          fo.write(str(request_pid) + "\n")
+
+  def _cpu_folder_list(self):
+    """Return list of folders for exclusive cpu cores."""
+    return [os.path.join(self.cpuset_path, "cpu" + str(cpu_id))
+            for cpu_id in self._cpu_id_list]
+
+  def _cpu_id_list(self):
+    """Extract IDs of available CPUs and return them as a list.
+
+    The first one will be always used for all non-exclusive processes.
+    :return: list[int]
+    """
+    cpu_list = []  # types: list[int]
+    with open(self.cpuset_path + "cpuset.cpus", "rt") as cpu_def:
+      for cpu_def_split in cpu_def.read().strip().split(","):
+        # IDs can be in form "0-4" or "0,1,2,3,4"
+        if "-" in cpu_def_split:
+          a, b = map(int, cpu_def_split.split("-"))
+          cpu_list.extend(range(a, b + 1)) # because cgroup's range is inclusive
+          continue
+        cpu_list.append(int(cpu_def_split))
+    return cpu_list
+
+  def _move_to_exclusive_cpu(self, pid):
+    """Try all exclusive CPUs and place the ``pid`` to the first available one.
+
+    :return: int, cpu_id of used CPU, -1 if placement was not possible
+    """
+    exclusive_cpu_list = self._cpu_id_list()[1:]
+    for exclusive_cpu in exclusive_cpu_list:
+      # gather tasks assigned to current exclusive CPU
+      task_path = os.path.join(self.cpuset_path, "cpu" + str(exclusive_cpu), "tasks")
+      with open(task_path, "rt") as fi:
+        task_list = fi.read().split()
+      if len(task_list) > 0:
+        continue  # skip occupied CPUs
+      return self._move_task(pid, exclusive_cpu)[1]
+    return -1
+
+  def _move_task(self, pid, cpu_id, cpu_mode="performance"):
+    """Move ``pid`` to ``cpu_id``.
+
+    cpu_mode can be "performance" or "powersave"
+    """
+    known_cpu_mode_list = ("performance", "powersave")
+    with open(os.path.join(self.cpuset_path, "cpu" + str(cpu_id), "tasks"), self.task_write_mode) as fo:
+      fo.write(str(pid) + "\n")
+    # set the core to `cpu_mode`
+    scaling_governor_file = "/sys/devices/system/cpu/cpu{:d}/cpufreq/scaling_governor".format(cpu_id)
+    if os.path.exists(scaling_governor_file):
+      if cpu_mode not in known_cpu_mode_list:
+        logger.warning("Cannot set CPU to mode \"{}\"! Known modes {!s}".format(
+          cpu_mode, known_cpu_mode_list))
+      else:
+        try:
+          with open(scaling_governor_file, self.task_write_mode) as fo:
+            fo.write(cpu_mode + "\n")  # default is "powersave"
+        except IOError as e:
+          # handle permission error
+          logger.error("Failed to write \"{}\" to {}".format(cpu_mode, scaling_governor_file))
+    return pid, cpu_id
+
+  def _prepare_folder(self, folder):
+    """If-Create folder and set group write permission."""
+    if not os.path.exists(folder):
+      os.mkdir(folder)
+      # make your life and testing easier and create mandatory files if they don't exist
+      mandatory_file_list = ("tasks", "cpuset.cpus")
+      for mandatory_file in mandatory_file_list:
+        file_path = os.path.join(folder, mandatory_file)
+        if not os.path.exists(file_path):
+          with open(file_path, "wb"):
+            pass  # touche
+    return folder
+
+
+def read_file(path, mode="rt"):
+  with open(path, mode) as fi:
+    return fi.read()
+
+
+def write_file(content, path, mode="wt"):
+  with open(path, mode) as fo:
+    fo.write(content)
\ No newline at end of file

slapos/manager/interface.py

+# coding: utf-8
+from zope.interface import Interface
+
+
+class IManager(Interface):
+  """Manager is called in every step of preparation of the computer."""
+
+  def __init__(config):
+    """Manager needs to know config for its functioning.
+
+    :param conf: dictionary-like object with full access to [slapos] section of the config file
+    """
+
+  def format(computer):
+    """Method called at `slapos node format` phase.
+
+    :param computer: slapos.format.Computer, currently formatted computer
+    """
+
+  def software(software):
+    """Method called at `slapos node software` phase.
+
+    :param software: slapos.grid.SlapObject.Software, currently processed software
+    """
+
+  def instance(partition):
+    """Method called at `slapos node instance` phase.
+
+    :param partition: slapos.grid.SlapObject.Partition, currently processed partition
+    """
+

slapos/tests/slapformat.py

@@ -30,6 +30,7 @@ import glob
 import logging
 import slapos.format
 import slapos.util
+import slapos.manager.cpuset
 import unittest
 
 import netaddr
@@ -41,6 +42,7 @@ import netifaces
 import os
 import pwd
 import time
+import mock
 
 USER_LIST = []
 GROUP_LIST = []
@@ -183,21 +185,6 @@ class SlaposUtilMock:
   def chownDirectory(*args, **kw):
     pass
 
-
-class CGroupManagerMock(slapos.format.CGroupManager):
-
-  short_name = 'cgroup_mock'
-  cpuset_path = "/tmp/cpuset/"
-  task_write_mode = "at"  # append insted of write tasks PIDs for the tests
-
-  def is_allowed(self):
-    """Always allowed."""
-    return True
-
-# update available managers with our partially-mocked version
-slapos.format.available_manager_list[CGroupManagerMock.short_name] = CGroupManagerMock
-
-
 class SlapformatMixin(unittest.TestCase):
   # keep big diffs
   maxDiff = None
@@ -668,23 +655,32 @@ class TestComputer(SlapformatMixin):
       self.fakeCallAndRead.external_command_list)
 
 
-class TestComputerWithCGroup(SlapformatMixin):
+class SlapGridPartitionMock:
+  def __init__(self, partition):
+    self.partition = partition
+    self.instance_path = partition.path
+
+
+class TestComputerWithCPUSet(SlapformatMixin):
+
+  cpuset_path = "/tmp/cpuset/"
+  task_write_mode = "at"  # append insted of write tasks PIDs for the tests
 
   def setUp(self):
-    super(TestComputerWithCGroup, self).setUp()
+    super(TestComputerWithCPUSet, self).setUp()
     self.restoreOs()
 
     if os.path.isdir("/tmp/slapgrid/"):
       shutil.rmtree("/tmp/slapgrid/")
     os.mkdir("/tmp/slapgrid/")
 
-    if os.path.isdir(CGroupManagerMock.cpuset_path):
-      shutil.rmtree(CGroupManagerMock.cpuset_path)
-    os.mkdir(CGroupManagerMock.cpuset_path)
+    if os.path.isdir(self.cpuset_path):
+      shutil.rmtree(self.cpuset_path)
+    os.mkdir(self.cpuset_path)
     file_write("0,1-3",
-               os.path.join(CGroupManagerMock.cpuset_path, "cpuset.cpus"))
+               os.path.join(self.cpuset_path, "cpuset.cpus"))
     file_write("\n".join(("1000", "1001", "1002", "")),
-               os.path.join(CGroupManagerMock.cpuset_path, "tasks"))
+               os.path.join(self.cpuset_path, "tasks"))
     self.cpu_list = [0, 1, 2, 3]
 
     global USER_LIST, INTERFACE_DICT
@@ -695,6 +691,13 @@ class TestComputerWithCGroup(SlapformatMixin):
       socket.AF_INET6: [
         {'addr': '2a01:e35:2e27::e59c', 'netmask': 'ffff:ffff:ffff:ffff::'}]
     }
+
+    from slapos.manager.cpuset import Manager
+    self.orig_cpuset_path = Manager.cpuset_path
+    self.orig_task_write_mode = Manager.task_write_mode
+    Manager.cpuset_path = self.cpuset_path
+    Manager.task_write_mode = self.task_write_mode
+
     self.computer = slapos.format.Computer('computer',
       software_user='testuser',
       instance_root='/tmp/slapgrid/instance_root',
@@ -705,47 +708,58 @@ class TestComputerWithCGroup(SlapformatMixin):
           slapos.format.Partition(
             'partition', '/tmp/slapgrid/instance_root/part1', slapos.format.User('testuser'), [], tap=None),
         ],
-      manager_list=(CGroupManagerMock.short_name, )
+      config={
+        "manager_list": "cpuset",
+        "power_user_list": "testuser"
+      }
     )
     # self.patchOs(self.logger)
 
   def tearDown(self):
     """Cleanup temporary test folders."""
-    super(TestComputerWithCGroup, self).tearDown()
+
+    from slapos.manager.cpuset import Manager
+    Manager.cpuset_path = self.orig_cpuset_path
+    Manager.task_write_mode = self.orig_task_write_mode
+
+    super(TestComputerWithCPUSet, self).tearDown()
     shutil.rmtree("/tmp/slapgrid/")
-    if CGroupManagerMock.cpuset_path.startswith("/tmp"):
-      shutil.rmtree(CGroupManagerMock.cpuset_path)
+    if self.cpuset_path.startswith("/tmp"):
+      shutil.rmtree(self.cpuset_path)
 
   def test_positive_cgroups(self):
     """Positive test of cgroups."""
     # Test parsing "cpuset.cpus" file
-    self.assertEqual(self.computer._manager_list[0]._cpu_list(), self.cpu_list)
+    self.assertEqual(self.computer._manager_list[0]._cpu_id_list(), self.cpu_list)
     # This should created per-cpu groups and move all tasks in CPU pool into cpu0
     self.computer.format(alter_network=False, alter_user=False)
     # Test files creation for exclusive CPUs
     for cpu_id in self.cpu_list:
-      cpu_n_path = os.path.join(CGroupManagerMock.cpuset_path, "cpu" + str(cpu_id))
+      cpu_n_path = os.path.join(self.cpuset_path, "cpu" + str(cpu_id))
       self.assertEqual(str(cpu_id), file_content(os.path.join(cpu_n_path, "cpuset.cpus")))
       self.assertEqual("1", file_content(os.path.join(cpu_n_path, "cpuset.cpu_exclusive")))
       if cpu_id > 0:
         self.assertEqual("", file_content(os.path.join(cpu_n_path, "tasks")))
+
+    # Simulate slapos instance call
+    self.computer._manager_list[0].instance(SlapGridPartitionMock(self.computer.partition_list[0]))
     # Test that format moved all PIDs from CPU pool into CPU0
-    tasks_at_cpu0 = file_content(os.path.join(CGroupManagerMock.cpuset_path, "cpu0", "tasks")).split()
+    tasks_at_cpu0 = file_content(os.path.join(self.cpuset_path, "cpu0", "tasks")).split()
     self.assertIn("1000", tasks_at_cpu0)
     self.assertIn("1001", tasks_at_cpu0)
     self.assertIn("1002", tasks_at_cpu0)
     # Simulate cgroup behaviour - empty tasks in the pool
-    file_write("", os.path.join(CGroupManagerMock.cpuset_path, "tasks"))
+    file_write("", os.path.join(self.cpuset_path, "tasks"))
     # test moving tasks from generic core to private core
     # request PID 1001 to be moved to its private CPU
     request_file_path = os.path.join(self.computer.partition_list[0].path,
-                                     CGroupManagerMock.cpu_exclusive_file)
+                                     self.cpu_exclusive_file)
     file_write("1001\n", request_file_path)
     # let format do the moving
     self.computer.update()
     # test if the moving suceeded into any provate CPUS (id>0)
     self.assertTrue(any("1001" in file_content(exclusive_task)
-                        for exclusive_task in glob.glob(os.path.join(CGroupManagerMock.cpuset_path, "cpu[1-9]", "tasks"))))
+                        for exclusive_task in glob.glob(os.path.join(self.cpuset_path, "cpu[1-9]", "tasks"))))
     # slapformat should remove successfully moved PIDs from the .slapos-cpu-exclusive file
     self.assertEqual("", file_content(request_file_path).strip())