feat(controller): improve the services api

Author: duanhongyi

rootfs/api/migrations/0001_initial.py

@@ -1,4 +1,4 @@
-# Generated by Django 3.2.14 on 2022-08-10 02:21
+# Generated by Django 4.1.7 on 2023-03-17 08:18
 
 import api.models.app
 import api.models.certificate
@@ -36,8 +36,8 @@ class Migration(migrations.Migration):
                 ('date_joined', models.DateTimeField(default=django.utils.timezone.now, verbose_name='date joined')),
                 ('id', models.BigIntegerField(primary_key=True, serialize=False, verbose_name='id')),
                 ('email', models.EmailField(max_length=254, unique=True, verbose_name='email address')),
-                ('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.Group', verbose_name='groups')),
-                ('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.Permission', verbose_name='user permissions')),
+                ('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_name='user_set', related_query_name='user', to='auth.group', verbose_name='groups')),
+                ('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='user_set', related_query_name='user', to='auth.permission', verbose_name='user permissions')),
             ],
             options={
                 'verbose_name': 'user',
@@ -202,6 +202,9 @@ class Migration(migrations.Migration):
                 ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                 ('created', models.DateTimeField(auto_now_add=True)),
                 ('updated', models.DateTimeField(auto_now=True)),
+                ('port', models.PositiveIntegerField(default=5000)),
+                ('protocol', models.TextField(default='TCP')),
+                ('target_port', models.PositiveIntegerField(default=5000)),
                 ('service_type', models.TextField()),
                 ('procfile_type', models.TextField()),
                 ('app', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='api.app')),

rootfs/api/models/service.py

@@ -1,6 +1,7 @@
 import logging
 
 from django.db import models
+from django.conf import settings
 from django.contrib.auth import get_user_model
 from api.exceptions import ServiceUnavailable
 from scheduler import KubeException
@@ -13,8 +14,11 @@
 class Service(AuditedModel):
     owner = models.ForeignKey(User, on_delete=models.PROTECT)
     app = models.ForeignKey('App', on_delete=models.CASCADE)
-    service_type = models.TextField(blank=False, null=False, unique=False)
-    procfile_type = models.TextField(blank=False, null=False, unique=False)
+    port = models.PositiveIntegerField(default=5000)
+    protocol = models.TextField(default="TCP")
+    target_port = models.PositiveIntegerField(default=5000)
+    service_type = models.TextField()
+    procfile_type = models.TextField()
 
     class Meta:
         get_latest_by = 'created'
@@ -24,22 +28,50 @@ class Meta:
     def __str__(self):
         return self._svc_name()
 
+    def _get_ips(self):
+        namespace = self._namespace()
+        svc_name = self._svc_name()
+        response = self._scheduler.svc.get(namespace, svc_name)
+        data = response.json()
+        cluster_ip = data['spec']['clusterIP']
+        if 'ingress' in data['status']['loadBalancer']:
+            external_ip = data['status']['loadBalancer']['ingress'][0]['ip']
+        else:
+            external_ip = None
+        return cluster_ip, external_ip
+
     def as_dict(self):
+        namespace = self._namespace()
+        svc_name = self._svc_name()
+        cluster_domain = settings.KUBERNETES_CLUSTER_DOMAIN
+        cluster_ip, external_ip = self._get_ips()
         return {
+            "port": self.port,
+            "domain": f"{svc_name}.{namespace}.svc.{cluster_domain}",
+            "protocol": self.protocol,
+            "cluster_ip": cluster_ip,
+            "external_ip": external_ip,
+            "target_port": self.target_port,
             "service_type": self.service_type,
-            "procfile_type": self.procfile_type
+            "procfile_type": self.procfile_type,
         }
 
-    def create(self, *args, **kwargs):  # noqa
-        # create required minimum service in k8s for the application
+    def create(self):
         namespace = self._namespace()
         svc_name = self._svc_name()
-        self.log('creating Service: {}'.format(svc_name), level=logging.DEBUG)
+        self.log('creating service: {}'.format(svc_name), level=logging.DEBUG)
         try:
             try:
                 self._scheduler.svc.get(namespace, svc_name)
             except KubeException:
-                self._scheduler.svc.create(namespace, svc_name, self.service_type)
+                self._scheduler.svc.create(
+                    namespace,
+                    svc_name,
+                    service_type=self.service_type,
+                    port=self.port,
+                    protocol=self.protocol,
+                    target_port=self.target_port,
+                )
         except KubeException as e:
             raise ServiceUnavailable('Kubernetes service could not be created') from e
 

rootfs/api/serializers.py

@@ -25,9 +25,10 @@
 
 User = get_user_model()
 logger = logging.getLogger(__name__)
-
 SVCTYPE_MATCH = re.compile(r'^(ClusterIP|LoadBalancer)$')
 SVCTYPE_MISMATCH_MSG = "The service type currently only supports ClusterIP and LoadBalancer"
+PROTOCOL_MATCH = re.compile(r'^(TCP|UDP|SCTP)$')
+PROTOCOL_MISMATCH_MSG = "Currently, the protocol only supports TCP, UDP, and SCTP"
 PROCTYPE_MATCH = re.compile(r'^(?P<type>[a-z0-9]+(\-[a-z0-9]+)*)$')
 PROCTYPE_MISMATCH_MSG = "Process types can only contain lowercase alphanumeric characters"
 MEMLIMIT_MATCH = re.compile(r'^(?P<mem>([1-9][0-9]*[mgMG]))$', re.IGNORECASE)
@@ -513,15 +514,40 @@ class ServiceSerializer(serializers.ModelSerializer):
 
     app = serializers.SlugRelatedField(slug_field='id', queryset=models.app.App.objects.all())
     owner = serializers.ReadOnlyField(source='owner.username')
-    service_type = serializers.CharField(allow_blank=False, allow_null=False, required=True)
-    procfile_type = serializers.CharField(allow_blank=False, allow_null=False, required=True)
+    port = serializers.IntegerField(default=5000)
+    protocol = serializers.CharField(default="TCP")
+    target_port = serializers.IntegerField(default=5000)
+    service_type = serializers.CharField(required=True)
+    procfile_type = serializers.CharField(required=True)
 
     class Meta:
         """Metadata options for a :class:`ServiceSerializer`."""
         model = models.service.Service
         fields = ['owner', 'created', 'updated', 'app', 'service_type', 'procfile_type']
         read_only_fields = ['uuid']
 
+    @staticmethod
+    def validate_port(value):
+        if not str(value).isnumeric():
+            raise serializers.ValidationError('port can only be a numeric value')
+        elif int(value) not in range(1, 65536):
+            raise serializers.ValidationError('port needs to be between 1 and 65535')
+        return value
+
+    @staticmethod
+    def validate_protocol(value):
+        if not re.match(PROTOCOL_MATCH, value):
+            raise serializers.ValidationError(PROTOCOL_MISMATCH_MSG)
+        return value
+
+    @staticmethod
+    def validate_target_port(value):
+        if not str(value).isnumeric():
+            raise serializers.ValidationError('target port can only be a numeric value')
+        elif int(value) not in range(1, 65536):
+            raise serializers.ValidationError('target port needs to be between 1 and 65535')
+        return value
+
     @staticmethod
     def validate_service_type(value):
         if not re.match(SVCTYPE_MATCH, value):

rootfs/api/settings/production.py

@@ -253,16 +253,16 @@
 # the k8s namespace in which the controller and workflow were installed.
 WORKFLOW_NAMESPACE = os.environ.get('WORKFLOW_NAMESPACE', 'drycc')
 
+# kubernetes cluster domain
+KUBERNETES_CLUSTER_DOMAIN = os.environ.get("KUBERNETES_CLUSTER_DOMAIN", "cluster.local")
 # default scheduler settings
 SCHEDULER_MODULE = 'scheduler'
 SCHEDULER_URL = "https://{}:{}".format(
     # accessing the k8s api server by IP address rather than hostname avoids
     # intermittent DNS errors
     os.environ.get(
         'KUBERNETES_SERVICE_HOST',
-        'kubernetes.default.svc.{}'.format(os.environ.get(
-            "KUBERNETES_CLUSTER_DOMAIN", "cluster.local"
-        ))
+        'kubernetes.default.svc.{}'.format(KUBERNETES_CLUSTER_DOMAIN)
     ),
     os.environ.get('KUBERNETES_SERVICE_PORT', '443')
 )

rootfs/api/tests/test_services.py

@@ -34,44 +34,71 @@ def test_service_basic_ops(self):
         # create 1st service
         response = self.client.post(
             '/v2/apps/{}/services'.format(app_id),
-            {'service_type': 'ClusterIP', 'procfile_type': 'test'}
+            {
+                'port': 5000,
+                'protocol': 'UDP',
+                'target_port': 5000,
+                'service_type': 'ClusterIP',
+                'procfile_type': 'test'
+            }
         )
         self.assertEqual(response.status_code, 201, response.data)
         # list 1st service
         response = self.client.get('/v2/apps/{}/services'.format(app_id))
         self.assertEqual(response.status_code, 200, response.data)
         self.assertEqual(len(response.data['services']), 1)
         expected1 = {
+            'port': 5000,
+            'protocol': 'UDP',
+            'target_port': 5000,
             'service_type': 'ClusterIP',
             'procfile_type': 'test'
         }
         self.assertDictContainsSubset(expected1, response.data['services'][0])
         # update 1st service
         response = self.client.post(
             '/v2/apps/{}/services'.format(app_id),
-            {'service_type': 'LoadBalancer', 'procfile_type': 'test'}
+            {
+                'port': 5000,
+                'protocol': 'UDP',
+                'target_port': 5000,
+                'service_type': 'LoadBalancer',
+                'procfile_type': 'test'
+            }
         )
         self.assertEqual(response.status_code, 201, response.data)
         # list 1st service and get new value
         response = self.client.get('/v2/apps/{}/services'.format(app_id))
         self.assertEqual(response.status_code, 200, response.data)
         self.assertEqual(len(response.data['services']), 1)
         expected1 = {
+            'port': 5000,
+            'protocol': 'UDP',
+            'target_port': 5000,
             'service_type': 'LoadBalancer',
             'procfile_type': 'test'
         }
         self.assertDictContainsSubset(expected1, response.data['services'][0])
         # create 2nd service
         response = self.client.post(
             '/v2/apps/{}/services'.format(app_id),
-            {'service_type': 'ClusterIP', 'procfile_type': 'test2'}
+            {
+                'port': 5000,
+                'protocol': 'UDP',
+                'target_port': 5000,
+                'service_type': 'ClusterIP',
+                'procfile_type': 'test2'
+            }
         )
         self.assertEqual(response.status_code, 201, response.data)
         # list two services
         response = self.client.get('/v2/apps/{}/services'.format(app_id))
         self.assertEqual(response.status_code, 200, response.data)
         self.assertEqual(len(response.data['services']), 2)
         expected2 = {
+            'port': 5000,
+            'protocol': 'UDP',
+            'target_port': 5000,
             'service_type': 'ClusterIP',
             'procfile_type': 'test2'
         }

rootfs/api/views.py

@@ -423,6 +423,9 @@ def list(self, *args, **kwargs):
     def create_or_update(self, request, **kwargs):
         svt = self.get_serializer().validate_service_type(request.data.get('service_type'))
         pft = self.get_serializer().validate_procfile_type(request.data.get('procfile_type'))
+        port = self.get_serializer().validate_port(request.data.get('port'))
+        protocol = self.get_serializer().validate_protocol(request.data.get('protocol'))
+        target_port = self.get_serializer().validate_target_port(request.data.get('target_port'))
         app = self.get_app()
         svc = app.service_set.filter(procfile_type=pft).first()
         if svc:
@@ -433,7 +436,14 @@ def create_or_update(self, request, **kwargs):
                 svc.save()
         else:
             svc = models.service.Service.objects.create(
-                owner=app.owner, app=app, service_type=svt, procfile_type=pft)
+                owner=app.owner,
+                app=app,
+                service_type=svt,
+                procfile_type=pft,
+                port=port,
+                protocol=protocol,
+                target_port=target_port,
+            )
         return Response(status=status.HTTP_201_CREATED)
 
     def delete(self, request, **kwargs):

rootfs/scheduler/mock.py

@@ -702,7 +702,15 @@ def post(request, context):
         namespace = re.sub(r'/apis/.+/.+/namespaces/', '', namespace)
         namespace = namespace.split('/')[0]
         data['metadata']['namespace'] = namespace
-
+    if resource_type in ["services"]:
+        data['spec']["clusterIP"] = "10.1.9.101"
+        data['status'] = {
+            'loadBalancer': {
+                'ingress': [
+                    {'ip': "47.98.100.101"}
+                ]
+            }
+        }
     # Handle RC / RS / Deployments
     if resource_type in ['replicationcontrollers', 'replicasets', 'deployments']:
         data['status'] = {

rootfs/scheduler/resources/service.py

@@ -27,7 +27,7 @@ def get(self, namespace, name=None, **kwargs):
 
         return response
 
-    def create(self, namespace, name, type="ClusterIP", **kwargs):
+    def create(self, namespace, name, **kwargs):
         # Ports and app type will be overwritten as required
         manifest = {
             'kind': 'Service',
@@ -41,12 +41,12 @@ def create(self, namespace, name, type="ClusterIP", **kwargs):
                 'annotations': {}
             },
             'spec': {
-                'type': type,
+                'type': kwargs.get("type", "ClusterIP"),
                 'ports': [{
-                    'name': 'http',
-                    'port': 80,
-                    'targetPort': 5000,
-                    'protocol': 'TCP'
+                    'name': name,
+                    'port': kwargs.get("port", 80),
+                    'targetPort': kwargs.get("target_port", 5000),
+                    'protocol': kwargs.get("protocol", "TCP"),
                 }],
                 'selector': {
                     'app': name,