feat(controller): volume support nfs

Author: duanhongyi

rootfs/api/migrations/0002_volume_parameters_volume_type_alter_volume_path_and_more.py

@@ -0,0 +1,33 @@
+# Generated by Django 4.2.10 on 2024-02-23 03:23
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='volume',
+            name='parameters',
+            field=models.JSONField(default=dict),
+        ),
+        migrations.AddField(
+            model_name='volume',
+            name='type',
+            field=models.CharField(choices=[('csi', 'container storage interface'), ('nfs', 'network file system')], default='csi'),
+        ),
+        migrations.AlterField(
+            model_name='volume',
+            name='path',
+            field=models.JSONField(default=dict),
+        ),
+        migrations.AlterField(
+            model_name='volume',
+            name='size',
+            field=models.CharField(default='0G', max_length=128),
+        ),
+    ]

rootfs/api/models/app.py

@@ -1076,14 +1076,14 @@ def _get_volumes_and_mounts(self, process_type, volumes):
         k8s_volumes, k8s_volume_mounts = [], []
         if volumes:
             for volume in volumes:
-                k8s_volumes.append({
-                    "name": volume.name,
-                    "claimName": volume.name
-                })
-                k8s_volume_mounts.append({
-                    "name": volume.name,
-                    "mount_path": volume.path.get(process_type)
-                })
+                k8s_volume = {"name": volume.name}
+                if volume.type == "csi":
+                    k8s_volume.update({"persistentVolumeClaim": {"claimName": volume.name}})
+                else:
+                    k8s_volume.update(volume.parameters)
+                k8s_volumes.append(k8s_volume)
+                k8s_volume_mounts.append(
+                    {"name": volume.name, "mountPath": volume.path.get(process_type)})
         k8s_volumes.extend(json.loads(settings.KUBERNETES_POD_DEFAULT_VOLUMES))
         k8s_volume_mounts.extend(json.loads(settings.KUBERNETES_POD_DEFAULT_VOLUME_MOUNTS))
         return k8s_volumes, k8s_volume_mounts

rootfs/api/models/volume.py

@@ -12,16 +12,22 @@
 
 
 class Volume(UuidAuditedModel):
+    TYPE_CHOICES = (
+        ("csi", "container storage interface"),
+        ("nfs", "network file system"),
+    )
     owner = models.ForeignKey(User, on_delete=models.PROTECT)
     app = models.ForeignKey('App', on_delete=models.CASCADE)
     name = models.CharField(max_length=63, validators=[validate_label])
-    size = models.CharField(max_length=128)
-    path = models.JSONField(default=dict, blank=True)
+    size = models.CharField(default='0G', max_length=128)
+    path = models.JSONField(default=dict)
+    type = models.CharField(default=TYPE_CHOICES[0][0], choices=TYPE_CHOICES)
+    parameters = models.JSONField(default=dict)
 
     @transaction.atomic
     def save(self, *args, **kwargs):
         # Attach volume, updates k8s
-        if self.created == self.updated:
+        if self.type == "csi" and self.created == self.updated:
             self._create_pvc()
         # Save to DB
         return super(Volume, self).save(*args, **kwargs)
@@ -31,26 +37,30 @@ def delete(self, *args, **kwargs):
         if self.path:
             raise DryccException("the volume is not unmounted")
         # Deatch volume, updates k8s
-        self._delete_pvc()
+        if self.type == "csi":
+            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.patch(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
+        if self.type == "csi":
+            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._format_size(self.size),
+                    "storage_class": settings.DRYCC_APP_STORAGE_CLASS,
+                }
+                self._scheduler.pvc.patch(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
+        else:
+            raise DryccException(f'{self.type} volume is not support expand.')
 
     def log(self, message, level=logging.INFO):
         """Logs a message in the context of this service.
@@ -70,15 +80,15 @@ def to_measurements(self, timestamp: float):
             "name": self.name,
             "type": "volume",
             "unit": "bytes",
-            "usage": unit_to_bytes(self.size),
+            "usage": unit_to_bytes(self.size) if self.type == "csi" else 0,
             "timestamp": int(timestamp)
         }]
 
     def __str__(self):
         return self.name
 
     @staticmethod
-    def _get_size(size):
+    def _format_size(size):
         """ Format volume limit value """
         if size[-2:-1].isalpha() and size[-1].isalpha():
             size = size[:-1]
@@ -97,7 +107,7 @@ def _create_pvc(self):
             logger.info(e)
             try:
                 kwargs = {
-                    "size": self._get_size(self.size),
+                    "size": self._format_size(self.size),
                     "storage_class": settings.DRYCC_APP_STORAGE_CLASS,
                 }
                 self._scheduler.pvc.create(self.app.id, self.name, **kwargs)

rootfs/api/schemas/volumes.py

@@ -0,0 +1,15 @@
+SCHEMA = {
+    "$schema": "http://json-schema.org/schema#",
+    "type": "object",
+    "properties": {
+        "nfs": {
+            "type": "object",
+            "properties": {
+                "server": {"type": "string"},
+                "path": {"type": "string"},
+                "readOnly": {"type": "boolean"},
+            },
+            "required": ["server", "path", "readOnly"],
+        },
+    },
+}

rootfs/api/serializers.py

@@ -18,28 +18,50 @@
 from api import models
 from api.exceptions import DryccException
 from api.schemas.rules import SCHEMA as RULES_SCHEMA
+from api.schemas.volumes import SCHEMA as VOLUMES_SCHEMA
 from api.schemas.autoscale import SCHEMA as AUTOSCALE_SCHEMA
 from api.schemas.healthcheck import SCHEMA as HEALTHCHECK_SCHEMA
 
 
 User = get_user_model()
 logger = logging.getLogger(__name__)
 SERVICE_PROTOCOL_MATCH = re.compile(r'^(TCP|UDP|SCTP)$')
-SERVICE_PROTOCOL_MISMATCH_MSG = "the service protocol only supports TCP, UDP, and SCTP"
+SERVICE_PROTOCOL_MISMATCH_MSG = (
+    "the service protocol only supports: %s" % SERVICE_PROTOCOL_MATCH.pattern)
 GATEWAY_PROTOCOL_MATCH = re.compile(r'^(HTTP|HTTPS|TCP|TLS|UDP)$')
-GATEWAY_PROTOCOL_MISMATCH_MSG = "the gateway protocol only supports HTTP, HTTPS, TCP, TLS and UDP"
+GATEWAY_PROTOCOL_MISMATCH_MSG = (
+    "the gateway protocol only supports: %s" % GATEWAY_PROTOCOL_MATCH.pattern)
 ROUTE_PROTOCOL_MATCH = re.compile(r'^(HTTPRoute|TCPRoute|UDPRoute|TLSRoute)$')
-ROUTE_PROTOCOL_MISMATCH_MSG = "the route kind only supports HTTPRoute, TCPRoute, UDPRoute, and TLSRoute"  # noqa
+ROUTE_PROTOCOL_MISMATCH_MSG = (
+    "the route kind only supports: %s" % ROUTE_PROTOCOL_MATCH.pattern)
 PROCTYPE_MATCH = re.compile(r'^(?P<type>[a-z0-9]+(\-[a-z0-9]+)*)(?<!-canary)$')
-PROCTYPE_MISMATCH_MSG = "Process types can only contain lowercase alphanumeric characters"
+PROCTYPE_MISMATCH_MSG = "Process types can only supports: %s" % PROCTYPE_MATCH.pattern
 MEMLIMIT_MATCH = re.compile(r'^(?P<mem>([1-9][0-9]*[mgMG]))$', re.IGNORECASE)
+MEMLIMIT_MISMATCH_MSG = (
+    "Memory limit format: <number><unit>, "
+    "where unit = M or G"
+)
 CPUSHARE_MATCH = re.compile(r'^(?P<cpu>([-+]?[1-9][0-9]*[m]?))$')
+CPUSHARE_MISMATCH_MSG = "CPU limit format: <value>, where value must be a numeric"
 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)
+CONFIGKEY_MISMATCH_MSG = (
+    "Config keys must start with a letter or underscore and "
+    "only contain [A-z0-9_]"
+)
+
 TERMINATION_GRACE_PERIOD_MATCH = re.compile(r'^[0-9]*$')
+TERMINATION_GRACE_PERIOD_MISMATCH_MSG = (
+    "Termination Grace Period format: %s" % TERMINATION_GRACE_PERIOD_MATCH.pattern)
+VOLUME_TYPE_MATCH = re.compile(r'^(csi|nfs)$')
+VOLUME_TYPE_MISMATCH_MSG = "Volume type pattern: %s" % VOLUME_TYPE_MATCH.pattern
 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$')
+VOLUME_SIZE_MISMATCH_MSG = (
+    "Volume size limit format: <number><unit> or <number><unit>/<number><unit>, "
+    "where unit = G, range: %sG~%sG"
+) % (settings.KUBERNETES_LIMITS_MAX_VOLUME, settings.KUBERNETES_LIMITS_MIN_VOLUME)
+VOLUME_PATH_MATCH = re.compile(r'^\/(\w+\/?)+$', re.IGNORECASE)
+METRIC_EVERY_MATCH = re.compile(r'^[1-9][0-9]*m$')
 
 
 class JSONFieldSerializer(serializers.JSONField):
@@ -179,9 +201,7 @@ def validate_values(data):
                 continue
 
             if not re.match(CONFIGKEY_MATCH, key):
-                raise serializers.ValidationError(
-                    "Config keys must start with a letter or underscore and "
-                    "only contain [A-z0-9_]")
+                raise serializers.ValidationError(CONFIGKEY_MISMATCH_MSG)
 
             # Validate PORT
             if key == 'PORT':
@@ -231,9 +251,7 @@ def validate_memory(data):
                 raise serializers.ValidationError(PROCTYPE_MISMATCH_MSG)
 
             if not re.match(MEMLIMIT_MATCH, str(value)):
-                raise serializers.ValidationError(
-                    "Memory limit format: <number><unit>, "
-                    "where unit = M or G")
+                raise serializers.ValidationError(MEMLIMIT_MISMATCH_MSG)
             range_error = "Memory setting is not in allowed range: %sM~%sM" % (
                 min_memory, max_memory)
             memory_size = int(value[:-1]) * 1024 if value.endswith("G") else int(value[:-1])
@@ -255,8 +273,7 @@ def validate_cpu(data):
 
             shares = re.match(CPUSHARE_MATCH, str(value))
             if not shares:
-                raise serializers.ValidationError(
-                    "CPU limit format: <value>, where value must be a numeric")
+                raise serializers.ValidationError(CPUSHARE_MISMATCH_MSG)
             range_error = "CPU setting is not in allowed range: %sm~%sm" % (
                 min_cpu, max_cpu)
             cpu_size = int(value) * 1000 if value.isdigit() else int(value[:-1])
@@ -322,9 +339,7 @@ def validate_registry(data):
                 continue
 
             if not re.match(CONFIGKEY_MATCH, key):
-                raise serializers.ValidationError(
-                    "Config keys must start with a letter or underscore and "
-                    "only contain [A-z0-9_]")
+                raise serializers.ValidationError(CONFIGKEY_MISMATCH_MSG)
 
         return data
 
@@ -560,28 +575,27 @@ class VolumeSerializer(serializers.ModelSerializer):
     app = serializers.SlugRelatedField(slug_field='id', queryset=models.app.App.objects.all())
     owner = serializers.ReadOnlyField(source='owner.username')
     name = serializers.CharField()
-    size = serializers.CharField()
+    size = serializers.CharField(required=False)
     path = JSONFieldSerializer(required=False, binary=True)
+    type = serializers.CharField(required=False)
+    parameters = serializers.JSONField(required=False)
 
     class Meta:
         """Metadata options for a :class:`AppVolumeSerializer`."""
         model = models.volume.Volume
         fields = '__all__'
 
-    @staticmethod
-    def validate_size(data):
+    def validate_size(self, data):
+        # check size format
         if not re.match(VOLUME_SIZE_MATCH, data):
-            raise serializers.ValidationError(
-                "Volume size limit format: <number><unit> or <number><unit>/<number><unit>, "
-                "where unit = G")
+            raise serializers.ValidationError(VOLUME_SIZE_MISMATCH_MSG)
+        # check volume size
+        volume_size = int(data[:-1])
         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: %sG~%sG" % (
-            min_volume, max_volume)
-        volume_size = int(data[:-1])
         if volume_size < min_volume or volume_size > max_volume:
-            raise serializers.ValidationError(range_error)
+            raise serializers.ValidationError(VOLUME_SIZE_MISMATCH_MSG)
         return data.upper()
 
     @staticmethod
@@ -595,7 +609,7 @@ def validate_path(data):
                 new_data[key] = value
                 continue
 
-            if not re.match(VOLUME_PATH, str(value)):
+            if not re.match(VOLUME_PATH_MATCH, str(value)):
                 raise serializers.ValidationError(
                     "Volume path format: /path")
             if value.endswith("/"):
@@ -604,6 +618,23 @@ def validate_path(data):
         logger.debug(f"mount validate_path new_data: {new_data}")
         return new_data
 
+    def validate_type(self, data):
+        if not re.match(VOLUME_TYPE_MATCH, data):
+            raise serializers.ValidationError(VOLUME_TYPE_MISMATCH_MSG)
+        elif data != "csi" and not self.initial_data.get("parameters", None):
+            raise serializers.ValidationError(
+                "parameters cannot be empty when the type is not csi.")
+        return data
+
+    @staticmethod
+    def validate_parameters(data):
+        try:
+            jsonschema.validate(data, VOLUMES_SCHEMA)
+        except jsonschema.ValidationError as e:
+            raise serializers.ValidationError(
+                "could not validate {}: {}".format(data, e.message))
+        return data
+
 
 class ResourceSerializer(serializers.ModelSerializer):
     """Serialize a :class:`~api.models.resource.Resource` model."""
@@ -641,9 +672,9 @@ class MetricSerializer(serializers.Serializer):
     every = serializers.CharField(max_length=50, required=False, default='5m')
 
     def validate(self, attrs):
-        if not re.match(METRIC_EVERY, attrs["every"]):
+        if not re.match(METRIC_EVERY_MATCH, attrs["every"]):
             raise serializers.ValidationError(
-                "The format of every is:%s" % METRIC_EVERY.pattern
+                "The format of every is:%s" % METRIC_EVERY_MATCH.pattern
             )
         interval = attrs.get("stop") - attrs.get("start")
         if interval < 0 or interval > 3600 * 24:

rootfs/api/tasks.py

@@ -101,6 +101,7 @@ def mount_app(app, user, volume):
         app.mount(user, volume)
         volume.save()
     except Exception as e:
+        print(e)
         signals.got_request_exception.send(sender=task_id)
         raise e
     else: