fix(scheduler): set env, limits, tags for run containers

Author: mboersma

rootfs/api/models/container.py

@@ -134,6 +134,10 @@ def run(self, command):
             raise EnvironmentError('No build associated with this release '
                                    'to run this command')
         image = self.release.image
+        kwargs = {'memory': self.release.config.memory,
+                  'cpu': self.release.config.cpu,
+                  'tags': self.release.config.tags,
+                  'envs': self.release.config.values}
         entrypoint = '/bin/bash'
         # if this is a procfile-based app, switch the entrypoint to slugrunner's default
         # FIXME: remove slugrunner's hardcoded entrypoint
@@ -145,7 +149,7 @@ def run(self, command):
         else:
             command = "-c '{}'".format(command)
         try:
-            rc, output = self._scheduler.run(self.job_id, image, entrypoint, command)
+            rc, output = self._scheduler.run(self.job_id, image, entrypoint, command, **kwargs)
             return rc, output
         except Exception as e:
             err = '{} (run): {}'.format(self.job_id, e)

rootfs/scheduler/__init__.py

@@ -446,7 +446,7 @@ def destroy(self, namespace):
             except KubeException:
                 break
 
-    def run(self, name, image, entrypoint, command):
+    def run(self, name, image, entrypoint, command, **kwargs):
         """Run a one-off command."""
         logger.debug('run {}, img {}, entrypoint {}, cmd "{}"'.format(
             name, image, entrypoint, command))
@@ -470,8 +470,12 @@ def run(self, name, image, entrypoint, command):
             args = [command[1:-1]]
 
         js_template = json.loads(template)
-        js_template['spec']['containers'][0]['command'] = [entrypoint]
-        js_template['spec']['containers'][0]['args'] = args
+        containers = js_template['spec']['containers'][0]
+        containers['command'] = [entrypoint]
+        containers['args'] = args
+
+        self._set_environment(containers, **kwargs)
+
         url = self._api("/namespaces/{}/pods", appname)
         resp = self.session.post(url, json=js_template)
         if unhealthy(resp.status_code):
@@ -508,6 +512,39 @@ def run(self, name, image, entrypoint, command):
             return err_code, data
         return 0, data
 
+    def _set_environment(self, json_data, **kwargs):
+        app_type = kwargs.get('app_type')
+        mem = kwargs.get('memory', {}).get(app_type)
+        cpu = kwargs.get('cpu', {}).get(app_type)
+        env = kwargs.get('envs', {})
+
+        if env:
+            for key, value in env.items():
+                json_data["env"].append({
+                    "name": key,
+                    "value": str(value)
+                })
+
+        # Inject debugging if workflow is in debug mode
+        if os.environ.get("DEBUG", False):
+            json_data["env"].append({
+                "name": "DEBUG",
+                "value": "1"
+            })
+
+        if mem or cpu:
+            json_data["resources"] = {"limits": {}}
+
+        if mem:
+            if mem[-2:-1].isalpha() and mem[-1].isalpha():
+                mem = mem[:-1]
+
+            mem = mem + "i"
+            json_data["resources"]["limits"]["memory"] = mem
+
+        if cpu:
+            json_data["resources"]["limits"]["cpu"] = cpu
+
     def state(self, name):
         """Display the state of a container."""
         try:
@@ -896,36 +933,8 @@ def _create_rc(self, namespace, name, image, command, **kwargs):  # noqa
         containers[0]['args'] = args
         loc = locals().copy()
         loc.update(re.match(MATCH, container_fullname).groupdict())
-        mem = kwargs.get('memory', {}).get(app_type)
-        cpu = kwargs.get('cpu', {}).get(app_type)
-        env = kwargs.get('envs', {})
 
-        if env:
-            for key, value in env.items():
-                containers[0]["env"].append({
-                    "name": key,
-                    "value": str(value)
-                })
-
-        # Inject debugging if workflow is in debug mode
-        if os.environ.get("DEBUG", False):
-            containers[0]["env"].append({
-                "name": "DEBUG",
-                "value": "1"
-            })
-
-        if mem or cpu:
-            containers[0]["resources"] = {"limits": {}}
-
-        if mem:
-            if mem[-2:-1].isalpha() and mem[-1].isalpha():
-                mem = mem[:-1]
-
-            mem = mem+"i"
-            containers[0]["resources"]["limits"]["memory"] = mem
-
-        if cpu:
-            containers[0]["resources"]["limits"]["cpu"] = cpu
+        self._set_environment(containers[0], **kwargs)
 
         # add in healtchecks
         if kwargs.get('healthcheck'):

rootfs/scheduler/abstract.py

@@ -15,7 +15,7 @@ def destroy(self, name):
         """Destroy a container."""
         raise NotImplementedError
 
-    def run(self, name, image, entrypoint, command):
+    def run(self, name, image, entrypoint, command, **kwargs):
         """Run a one-off command."""
         raise NotImplementedError
 

rootfs/scheduler/chaos.py

@@ -26,7 +26,7 @@ def destroy(self, name):
         else:
             super(ChaosSchedulerClient, self).destroy(name)
 
-    def run(self, name, image, entrypoint, command):
+    def run(self, name, image, entrypoint, command, **kwargs):
         """Run a one-off command."""
         if random.random() < CREATE_ERROR_RATE:
             raise RuntimeError('exit code 1')

rootfs/scheduler/mock.py

@@ -22,7 +22,7 @@ def destroy(self, name):
         """Destroy a container."""
         jobs.setdefault(name, {})['state'] = JobState.destroyed
 
-    def run(self, name, image, entrypoint, command):
+    def run(self, name, image, entrypoint, command, **kwargs):
         """Run a one-off command."""
         # dump input into a json object for testing purposes
         return 0, json.dumps({