Merge pull request #1513 from deis/resource-limits

Memory/CPU limits for application containers

Author: Matthew Fisher

client/deis.py

@@ -18,6 +18,7 @@
   config        manage environment variables that define app config
   domains       manage and assign domain names to your applications
   builds        manage builds created using `git push`
+  limits        manage resource limits for your application
   releases      manage releases of an application
 
   keys          manage ssh keys used for `git push` deployments
@@ -1345,6 +1346,162 @@ def domains_list(self, args):
         else:
             raise ResponseError(response)
 
+    def limits(self, args):
+        """
+        Valid commands for limits:
+
+        limits:list        list resource limits for an app
+        limits:set         set resource limits for an app
+        limits:unset       unset resource limits for an app
+
+        Use `deis help [command]` to learn more.
+        """
+        sys.argv[1] = 'limits:list'
+        args = docopt(self.limits_list.__doc__)
+        return self.limits_list(args)
+
+    def limits_list(self, args):
+        """
+        Lists resource limits for an application.
+
+        Usage: deis limits:list [options]
+
+        Options:
+          -a --app=<app>
+            the uniquely identifiable name of the application.
+        """
+        app = args.get('--app')
+        if not app:
+            app = self._session.app
+        response = self._dispatch('get', "/api/apps/{}/limits".format(app))
+        if response.status_code == requests.codes.ok:  # @UndefinedVariable
+            self._print_limits(app, response.json())
+        else:
+            raise ResponseError(response)
+
+    def limits_set(self, args):
+        """
+        Sets resource limits for an application.
+
+        A resource limit is a finite resource within a container which we can apply
+        restrictions to either through the scheduler or through the Docker API. This limit
+        is applied to each individual container, so setting a memory limit of 1G for an
+        application means that each container gets 1G of memory.
+
+        Usage: deis limits:set [options] <type>=<limit>...
+
+        Arguments:
+          <type>
+            the process type as defined in your Procfile, such as 'web' or 'worker'.
+            Note that Dockerfile apps have a default 'cmd' process type.
+          <limit>
+            The limit to apply to the process type. By default, this is set to --memory.
+            You can only set one type of limit per call.
+
+            With --memory, units are represented in Bytes (B), Kilobytes (K), Megabytes
+            (M), or Gigabytes (G). For example, `deis limit:set cmd=1G` will restrict all
+            "cmd" processes to a maximum of 1 Gigabyte of memory each.
+
+            With --cpu, units are represented in the number of cpu shares. For example,
+            `deis limit:set --cpu cmd=1024` will restrict all "cmd" processes to a
+            maximum of 1024 cpu shares.
+
+        Options:
+          -a --app=<app>
+            the uniquely identifiable name for the application.
+          -c --cpu
+            limits cpu shares.
+          -m --memory
+            limits memory. [default: true]
+        """
+        app = args.get('--app')
+        if not app:
+            app = self._session.app
+        body = {}
+        # see if cpu shares are being specified, otherwise default to memory
+        target = 'cpu' if args.get('--cpu') else 'memory'
+        body[target] = json.dumps(dictify(args['<type>=<limit>']))
+        sys.stdout.write('Applying limits... ')
+        sys.stdout.flush()
+        try:
+            progress = TextProgress()
+            progress.start()
+            response = self._dispatch('post', "/api/apps/{}/limits".format(app), json.dumps(body))
+        finally:
+            progress.cancel()
+            progress.join()
+        if response.status_code == requests.codes.created:  # @UndefinedVariable
+            version = response.headers['x-deis-release']
+            print("done, v{}\n".format(version))
+
+            self._print_limits(app, response.json())
+        else:
+            raise ResponseError(response)
+
+    def limits_unset(self, args):
+        """
+        Unsets resource limits for an application.
+
+        Usage: deis limits:unset [options] [--memory | --cpu] <type>...
+
+        Arguments:
+          <type>
+            the process type as defined in your Procfile, such as 'web' or 'worker'.
+            Note that Dockerfile apps have a default 'cmd' process type.
+
+        Options:
+          -a --app=<app>
+            the uniquely identifiable name for the application.
+          -c --cpu
+            limits cpu shares.
+          -m --memory
+            limits memory. [default: true]
+        """
+        app = args.get('--app')
+        if not app:
+            app = self._session.app
+        values = {}
+        for k in args.get('<type>'):
+            values[k] = None
+        body = {}
+        # see if cpu shares are being specified, otherwise default to memory
+        target = 'cpu' if args.get('--cpu') else 'memory'
+        body[target] = json.dumps(values)
+        sys.stdout.write('Applying limits... ')
+        sys.stdout.flush()
+        try:
+            progress = TextProgress()
+            progress.start()
+            response = self._dispatch('post', "/api/apps/{}/limits".format(app), json.dumps(body))
+        finally:
+            progress.cancel()
+            progress.join()
+        if response.status_code == requests.codes.created:  # @UndefinedVariable
+            version = response.headers['x-deis-release']
+            print("done, v{}\n".format(version))
+            self._print_limits(app, response.json())
+        else:
+            raise ResponseError(response)
+
+    def _print_limits(self, app, limit):
+        print("=== {} Limits".format(app))
+
+        def write(d):
+            items = d.items()
+            if len(items) == 0:
+                print('Unlimited')
+                return
+            keys = sorted(d)
+            width = max(map(len, keys)) + 5
+            for k in keys:
+                v = d[k]
+                print(("{k:<" + str(width) + "} {v}").format(**locals()))
+
+        print("\n--- Memory")
+        write(json.loads(limit.get('memory', '{}')))
+        print("\n--- CPU")
+        write(json.loads(limit.get('cpu', '{}')))
+
     def ps(self, args):
         """
         Valid commands for processes:

controller/api/models.py

@@ -301,10 +301,15 @@ def _command_announceable(self):
     @transition(field=state, source=INITIALIZED, target=CREATED)
     def create(self):
         image = self.release.image
+        kwargs = {}
+        if self.release.config.limit is not None:
+            kwargs = {'memory': self.release.config.limit.memory,
+                      'cpu': self.release.config.limit.cpu}
         self._scheduler.create(name=self._job_id,
                                image=image,
                                command=self._command,
-                               use_announcer=self._command_announceable())
+                               use_announcer=self._command_announceable(),
+                               **kwargs)
 
     @close_db_connections
     @transition(field=state,
@@ -327,10 +332,15 @@ def deploy(self, release):
         new_job_id = self._job_id
         image = self.release.image
         c_type = self.type
+        kwargs = {}
+        if self.release.config.limit is not None:
+            kwargs = {'memory': self.release.config.limit.memory,
+                      'cpu': self.release.config.limit.cpu}
         self._scheduler.create(name=new_job_id,
                                image=image,
                                command=self._command.format(**locals()),
-                               use_announcer=self._command_announceable())
+                               use_announcer=self._command_announceable(),
+                               **kwargs)
         self._scheduler.start(new_job_id, self._command_announceable())
         # destroy old container
         self._scheduler.destroy(old_job_id, self._command_announceable())
@@ -416,6 +426,28 @@ class Config(UuidAuditedModel):
     owner = models.ForeignKey(settings.AUTH_USER_MODEL)
     app = models.ForeignKey('App')
     values = JSONField(default='{}', blank=True)
+    limit = models.ForeignKey('Limit', null=True)
+
+    class Meta:
+        get_latest_by = 'created'
+        ordering = ['-created']
+        unique_together = (('app', 'uuid'),)
+
+    def __str__(self):
+        return "{}-{}".format(self.app.id, self.uuid[:7])
+
+
+@python_2_unicode_compatible
+class Limit(UuidAuditedModel):
+    """
+    Set of resource limits applied by the scheduler
+    during runtime execution of the Application.
+    """
+
+    owner = models.ForeignKey(settings.AUTH_USER_MODEL)
+    app = models.ForeignKey('App')
+    memory = JSONField(default='{}', blank=True)
+    cpu = JSONField(default='{}', blank=True)
 
     class Meta:
         get_latest_by = 'created'
@@ -507,12 +539,14 @@ def previous(self):
             prev_release = None
         return prev_release
 
-    def save(self, *args, **kwargs):
+    def save(self, *args, **kwargs):  # noqa
         if not self.summary:
             self.summary = ''
             prev_release = self.previous()
             # compare this build to the previous build
             old_build = prev_release.build if prev_release else None
+            old_config = prev_release.config if prev_release else None
+            old_limit = prev_release.config.limit if prev_release else None
             # if the build changed, log it and who pushed it
             if self.version == 1:
                 self.summary += "{} created initial release".format(self.app.owner)
@@ -521,10 +555,8 @@ def save(self, *args, **kwargs):
                     self.summary += "{} deployed {}".format(self.build.owner, self.build.sha[:7])
                 else:
                     self.summary += "{} deployed {}".format(self.build.owner, self.build.image)
-            # compare this config to the previous config
-            old_config = prev_release.config if prev_release else None
             # if the config data changed, log the dict diff
-            if self.config != old_config:
+            elif self.config != old_config:
                 dict1 = self.config.values
                 dict2 = old_config.values if old_config else {}
                 diff = dict_diff(dict1, dict2)
@@ -540,11 +572,25 @@ def save(self, *args, **kwargs):
                     if self.summary:
                         self.summary += ' and '
                     self.summary += "{} {}".format(self.config.owner, changes)
-                if not self.summary:
-                    if self.version == 1:
-                        self.summary = "{} created the initial release".format(self.owner)
-                    else:
-                        self.summary = "{} changed nothing".format(self.owner)
+            # if the limit changes, log the dict diff
+            elif self.config.limit != old_limit:
+                changes = []
+                old_mem = old_limit.memory if old_limit else {}
+                diff = dict_diff(self.limit.memory, old_mem)
+                if diff.get('added') or diff.get('changed') or diff.get('deleted'):
+                    changes.append('memory')
+                old_cpu = old_limit.cpu if old_limit else {}
+                diff = dict_diff(self.limit.cpu, old_cpu)
+                if diff.get('added') or diff.get('changed') or diff.get('deleted'):
+                    changes.append('cpu')
+                if changes:
+                    changes = 'changed limits for '+', '.join(changes)
+                    self.summary += "{} {}".format(self.config.owner, changes)
+            if not self.summary:
+                if self.version == 1:
+                    self.summary = "{} created the initial release".format(self.owner)
+                else:
+                    self.summary = "{} changed nothing".format(self.owner)
         super(Release, self).save(*args, **kwargs)
 
 

controller/api/serializers.py

@@ -14,6 +14,11 @@
 from api import utils
 
 
+PROCTYPE_MATCH = re.compile(r'^(?P<type>[a-z]+)')
+MEMLIMIT_MATCH = re.compile(r'^(?P<mem>[0-9]+[BbKkMmGg])$')
+CPUSHARE_MATCH = re.compile(r'^(?P<cpu>[0-9]+)$')
+
+
 class OwnerSlugRelatedField(serializers.SlugRelatedField):
     """Filter queries by owner as well as slug_field."""
 
@@ -102,6 +107,51 @@ class Meta:
         read_only_fields = ('uuid', 'created', 'updated')
 
 
+class LimitSerializer(serializers.ModelSerializer):
+    """Serialize a :class:`~api.models.Limit` model."""
+
+    owner = serializers.Field(source='owner.username')
+    app = serializers.SlugRelatedField(slug_field='id')
+    memory = serializers.ModelField(
+        model_field=models.Limit()._meta.get_field('memory'), required=False)
+    cpu = serializers.ModelField(
+        model_field=models.Limit()._meta.get_field('cpu'), required=False)
+
+    class Meta:
+        """Metadata options for a :class:`LimitSerializer`."""
+        model = models.Limit
+        read_only_fields = ('uuid', 'created', 'updated')
+
+    def validate_memory(self, attrs, source):
+        for k, v in attrs.get(source, {}).items():
+            if v is None:  # use NoneType to unset a value
+                continue
+            if not re.match(PROCTYPE_MATCH, k):
+                raise serializers.ValidationError("Process types can only contain [a-z]")
+            if not re.match(MEMLIMIT_MATCH, str(v)):
+                raise serializers.ValidationError(
+                    "Limit format: <number><unit>, where unit = B, K, M or G")
+        return attrs
+
+    def validate_cpu(self, attrs, source):
+        for k, v in attrs.get(source, {}).items():
+            if v is None:  # use NoneType to unset a value
+                continue
+            if not re.match(PROCTYPE_MATCH, k):
+                raise serializers.ValidationError("Process types can only contain [a-z]")
+            shares = re.match(CPUSHARE_MATCH, str(v))
+            if not shares:
+                raise serializers.ValidationError("CPU shares must be an integer")
+            for v in shares.groupdict().values():
+                try:
+                    i = int(v)
+                except ValueError:
+                    raise serializers.ValidationError("CPU shares must be an integer")
+                if i > 1024 or i < 0:
+                    raise serializers.ValidationError("CPU shares must be between 0 and 1024")
+        return attrs
+
+
 class ReleaseSerializer(serializers.ModelSerializer):
     """Serialize a :class:`~api.models.Release` model."""