feat(controller): add volume expand support

Author: duanhongyi

rootfs/api/models/volume.py

@@ -18,19 +18,11 @@ class Volume(UuidAuditedModel):
     size = models.CharField(max_length=128)
     path = models.JSONField(default=dict, blank=True)
 
-    class Meta:
-        get_latest_by = 'created'
-        unique_together = (('app', 'name'),)
-        ordering = ['-created']
-
-    def __str__(self):
-        return self.name
-
     @transaction.atomic
     def save(self, *args, **kwargs):
         # Attach volume, updates k8s
         if self.created == self.updated:
-            self.attach()
+            self._create_pvc()
         # Save to DB
         return super(Volume, self).save(*args, **kwargs)
 
@@ -39,10 +31,52 @@ def delete(self, *args, **kwargs):
         if self.path:
             raise DryccException("the volume is not unmounted")
         # Deatch volume, updates k8s
-        self.detach()
+        self._delete_pvc()
         # Delete from DB
         return super(Volume, self).delete(*args, **kwargs)
 
+    @transaction.atomic
+    def expand(self, size):
+        if unit_to_bytes(size) < unit_to_bytes(self.size):
+            raise DryccException('Shrink volume is not supported.')
+        self.size = size
+        self.save()
+        try:
+            kwargs = {
+                "size": self._get_size(self.size),
+                "storage_class": settings.DRYCC_APP_STORAGE_CLASS,
+            }
+            self._scheduler.pvc.put(self.app.id, self.name, **kwargs)
+        except KubeException as e:
+            msg = 'There was a problem expand the volume ' \
+                    '{} for {}'.format(self.name, self.app_id)
+            raise ServiceUnavailable(msg) from e
+
+    def log(self, message, level=logging.INFO):
+        """Logs a message in the context of this service.
+
+        This prefixes log messages with an application "tag" that the customized
+        drycc-logspout will be on the lookout for.  When it's seen, the message-- usually
+        an application event of some sort like releasing or scaling, will be considered
+        as "belonging" to the application instead of the controller and will be handled
+        accordingly.
+        """
+        logger.log(level, "[{}]: {}".format(self.id, message))
+
+    def to_measurements(self, timestamp: float):
+        return [{
+            "app_id": str(self.app_id),
+            "user_id": str(self.owner_id),
+            "name": self.name,
+            "type": "volume",
+            "unit": "bytes",
+            "usage": unit_to_bytes(self.size),
+            "timestamp": "%f" % timestamp
+        }]
+
+    def __str__(self):
+        return self.name
+
     @staticmethod
     def _get_size(size):
         """ Format volume limit value """
@@ -53,7 +87,7 @@ def _get_size(size):
             size = size.upper() + "i"
         return size
 
-    def attach(self):
+    def _create_pvc(self):
         try:
             self._scheduler.pvc.get(self.app.id, self.name)
             err = "Volume {} already exists in this namespace".format(self.name)  # noqa
@@ -72,7 +106,7 @@ def attach(self):
                       '{} for {}'.format(self.name, self.app_id)
                 raise ServiceUnavailable(msg) from e
 
-    def detach(self):
+    def _delete_pvc(self):
         try:
             # We raise an exception when a volume doesn't exist
             self._scheduler.pvc.get(self.app.id, self.name)
@@ -81,24 +115,7 @@ def detach(self):
             raise ServiceUnavailable("Could not delete volume {} for application \
                 {}".format(self.name, self.app_id)) from e  # noqa
 
-    def log(self, message, level=logging.INFO):
-        """Logs a message in the context of this service.
-
-        This prefixes log messages with an application "tag" that the customized
-        drycc-logspout will be on the lookout for.  When it's seen, the message-- usually
-        an application event of some sort like releasing or scaling, will be considered
-        as "belonging" to the application instead of the controller and will be handled
-        accordingly.
-        """
-        logger.log(level, "[{}]: {}".format(self.id, message))
-
-    def to_measurements(self, timestamp: float):
-        return [{
-            "app_id": str(self.app_id),
-            "user_id": str(self.owner_id),
-            "name": self.name,
-            "type": "volume",
-            "unit": "bytes",
-            "usage": unit_to_bytes(self.size),
-            "timestamp": "%f" % timestamp
-        }]
+    class Meta:
+        get_latest_by = 'created'
+        unique_together = (('app', 'name'),)
+        ordering = ['-created']

rootfs/api/serializers.py

@@ -33,7 +33,7 @@
 TAGVAL_MATCH = re.compile(r'^(?:[a-zA-Z\d][-\.\w]{0,61})?[a-zA-Z\d]$')
 CONFIGKEY_MATCH = re.compile(r'^[a-z_]+[a-z0-9_]*$', re.IGNORECASE)
 TERMINATION_GRACE_PERIOD_MATCH = re.compile(r'^[0-9]*$')
-VOLUME_SIZE_MATCH = re.compile(r'^(?P<volume>([1-9][0-9]*[mgMG]))$', re.IGNORECASE)
+VOLUME_SIZE_MATCH = re.compile(r'^(?P<volume>([1-9][0-9]*[gG]))$', re.IGNORECASE)
 VOLUME_PATH = re.compile(r'^\/(\w+\/?)+$', re.IGNORECASE)
 METRIC_EVERY = re.compile(r'^[1-9][0-9]*m$')
 
@@ -669,13 +669,13 @@ def validate_size(data):
         if not re.match(VOLUME_SIZE_MATCH, data):
             raise serializers.ValidationError(
                 "Volume size limit format: <number><unit> or <number><unit>/<number><unit>, "
-                "where unit = M or G")
+                "where unit = G")
         max_volume = settings.KUBERNETES_LIMITS_MAX_VOLUME
         # The minimum limit memory is equal to the memory allocated by default
         min_volume = settings.KUBERNETES_LIMITS_MIN_VOLUME
-        range_error = "Volume setting is not in allowed range: %sM~%sM" % (
+        range_error = "Volume setting is not in allowed range: %sG~%sG" % (
             min_volume, max_volume)
-        volume_size = int(data[:-1]) * 1024 if data.endswith("G") else int(data[:-1])
+        volume_size = int(data[:-1])
         if volume_size < min_volume or volume_size > max_volume:
             raise serializers.ValidationError(range_error)
         return data.upper()

rootfs/api/settings/production.py

@@ -353,10 +353,10 @@
 KUBERNETES_LIMITS_MIN_MEMORY = 128
 # Max Memory limit, units are represented in Megabytes(M)
 KUBERNETES_LIMITS_MAX_MEMORY = 128 * 1024
-# Minimum Memory limit, units are represented in Megabytes(M)
-KUBERNETES_LIMITS_MIN_VOLUME = 128
-# Max Memory limit, units are represented in Megabytes(M)
-KUBERNETES_LIMITS_MAX_VOLUME = 128 * 1024
+# Minimum Stroage Volume limit, units are represented in Gigabytes(G)
+KUBERNETES_LIMITS_MIN_VOLUME = 1
+# Max Stroage Volume limit, units are represented in Gigabytes(G)
+KUBERNETES_LIMITS_MAX_VOLUME = 1024 * 16
 
 # Default pod spec for application.
 KUBERNETES_POD_DEFAULT_RESOURCES = os.environ.get(

rootfs/api/tests/test_volume.py

@@ -39,7 +39,7 @@ def test_volumecreate(self, mock_requests):
         response = self.client.post(
             '/v2/apps/{}/volumes'.format(app_id),
             data={
-                'name': 'myvolume', 'size': '500M'
+                'name': 'myvolume', 'size': '500G'
             }
         )
         self.assertEqual(response.status_code, 201, response.data)
@@ -53,7 +53,7 @@ def test_volumecreate(self, mock_requests):
             'owner': self.user.username,
             'app': app_id,
             'name': 'myvolume',
-            'size': '500M'
+            'size': '500G'
         }
         self.assertDictContainsSubset(expected, response.data)
 
@@ -65,8 +65,8 @@ def test_volume_list_unmount(self, mock_requests):
         # create
         app_id = self.create_app()
         data = [
-            {'name': 'myvolume1', 'size': '500M'},
-            {'name': 'myvolume2', 'size': '500M'}
+            {'name': 'myvolume1', 'size': '500G'},
+            {'name': 'myvolume2', 'size': '500G'}
         ]
         for _ in data:
             self.client.post('/v2/apps/{}/volumes'.format(app_id), data=_)
@@ -76,6 +76,36 @@ def test_volume_list_unmount(self, mock_requests):
         expected = [res['name'] for res in response.data['results']]
         self.assertEqual(sorted([_['name'] for _ in data]), sorted(expected))
 
+    def test_volume_expand(self, mock_requests):
+        """
+        Test that volume is delete for a new app and that
+        volume can be updated using a PATCH
+        """
+        # create
+        app_id = self.create_app()
+        data = {'name': 'myvolume1', 'size': '500G'}
+        self.client.post('/v2/apps/{}/volumes'.format(app_id), data=data)
+
+        # Put
+        url = '/v2/apps/{app_id}/volumes/{volume}'.format(app_id=app_id,
+                                                          volume='myvolume1')
+        response = self.client.put(url, {'name': 'myvolume1', 'size': '100G'})
+        self.assertEqual(response.status_code, 400)
+
+        response = self.client.put(url, {'name': 'myvolume1', 'size': '1024G'})
+        self.assertEqual(response.status_code, 200)
+        # Fetch
+        url = '/v2/apps/{app_id}/volumes'.format(app_id=app_id)
+        response = self.client.get(url)
+        expected = {
+            'owner': self.user.username,
+            'app': app_id,
+            'name': 'myvolume1',
+            'size': '1024G'
+        }
+        assert len(response.data["results"]) == 1
+        self.assertDictContainsSubset(expected, response.data["results"][0])
+
     def test_volume_delete(self, mock_requests):
         """
         Test that volume is delete for a new app and that
@@ -84,8 +114,8 @@ def test_volume_delete(self, mock_requests):
         # create
         app_id = self.create_app()
         data = [
-            {'name': 'myvolume1', 'size': '500M'},
-            {'name': 'myvolume2', 'size': '500M'}
+            {'name': 'myvolume1', 'size': '500G'},
+            {'name': 'myvolume2', 'size': '500G'}
         ]
         for _ in data:
             self.client.post('/v2/apps/{}/volumes'.format(app_id), data=_)
@@ -108,7 +138,7 @@ def test_volume_mount(self, mock_requests):
         # create
         app_id = self.create_app()
         data = [
-            {'name': 'myvolume1', 'size': '500M'}
+            {'name': 'myvolume1', 'size': '500G'}
         ]
         for _ in data:
             self.client.post('/v2/apps/{}/volumes'.format(app_id), data=_)
@@ -125,7 +155,7 @@ def test_volume_unmount(self, mock_requests):
         # create
         app_id = self.create_app()
         data = [
-            {'name': 'myvolume1', 'size': '500M'}
+            {'name': 'myvolume1', 'size': '500G'}
         ]
         for _ in data:
             self.client.post('/v2/apps/{}/volumes'.format(app_id), data=_)
@@ -183,8 +213,8 @@ def test_measure_volumes(self, *args, **kwargs):
         # create
         app_id = self.create_app()
         data = [
-            {'name': 'myvolume1', 'size': '500M'},
-            {'name': 'myvolume2', 'size': '500M'}
+            {'name': 'myvolume1', 'size': '500G'},
+            {'name': 'myvolume2', 'size': '500G'}
         ]
         for _ in data:
             self.client.post('/v2/apps/{}/volumes'.format(app_id), data=_)

rootfs/api/urls.py

@@ -79,7 +79,7 @@
     url(r"^apps/(?P<id>{})/volumes/?$".format(settings.APP_URL_REGEX),
         views.AppVolumesViewSet.as_view({'get': 'list', 'post': 'create'})),
     url(r"^apps/(?P<id>{})/volumes/(?P<name>[-_\w]+)/?$".format(settings.APP_URL_REGEX),
-        views.AppVolumesViewSet.as_view({'delete': 'destroy'})),
+        views.AppVolumesViewSet.as_view({'put': 'expand', 'delete': 'destroy'})),
     url(r"^apps/(?P<id>{})/volumes/(?P<name>[-_\w]+)/path/?$".format(settings.APP_URL_REGEX),
         views.AppVolumeMountPathViewSet.as_view({'patch': 'path'})),
     # application resources

rootfs/api/views.py

@@ -680,6 +680,14 @@ class AppVolumesViewSet(ReleasableViewSet):
     model = models.volume.Volume
     serializer_class = serializers.VolumeSerializer
 
+    def expand(self, request, **kwargs):
+        volume = get_object_or_404(models.volume.Volume,
+                                   app__id=self.kwargs['id'],
+                                   name=self.kwargs['name'])
+        volume.expand(request.data['size'])
+        serializer = self.get_serializer(volume, many=False)
+        return Response(serializer.data)
+
     def destroy(self, request, **kwargs):
         volume = get_object_or_404(models.volume.Volume,
                                    app__id=self.kwargs['id'],

rootfs/scheduler/resources/pvc.py

@@ -1,3 +1,4 @@
+import json
 from scheduler.resources import Resource
 from scheduler.exceptions import KubeHTTPException
 
@@ -64,6 +65,16 @@ def create(self, namespace, name, **kwargs):
                 "create persistentvolumeclaim {}".format(namespace))
         return response
 
+    def put(self, namespace, name, **kwargs):
+        url = self.api('/namespaces/{}/persistentvolumeclaims/{}', namespace,
+                       name)
+        data = self.manifest(namespace, name, **kwargs)
+        response = self.http_put(url, json=data)
+        if self.unhealthy(response.status_code):
+            self.log(namespace, 'template used: {}'.format(json.dumps(data, indent=4)), 'DEBUG')  # noqa
+            raise KubeHTTPException(response, 'update HorizontalPodAutoscaler "{}"', name)
+        return response
+
     def delete(self, namespace, name):
         """
         Delete persistentvolumeclaim

rootfs/scheduler/resources/scale.py

@@ -1,3 +1,4 @@
+import json
 from scheduler.resources import Resource
 from scheduler.exceptions import KubeHTTPException
 
@@ -31,10 +32,9 @@ def update(self, namespace, name, replicas, target):
         url = self.api("/namespaces/{}/{}/{}/scale", namespace, resource_type, name)
         response = self.http_put(url, json=manifest)
         if self.unhealthy(response.status_code):
+            self.log(namespace, 'template used: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')  # noqa
             raise KubeHTTPException(
                 response,
                 'scale {} "{}" in Namespace "{}"', target['kind'], name, namespace
             )
-            self.log(namespace, 'template used: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')  # noqa
-
         return response