feat(pipeline): add dryccfile support

Author: duanhongyi

charts/controller/templates/controller-celery-deloyment.yaml

@@ -39,8 +39,7 @@ spec:
           - $(DRYCC_REDIS_ADDRS),$(DRYCC_CONTROLLER_API_SERVICE_HOST):$(DRYCC_CONTROLLER_API_SERVICE_PORT)
         {{- include "controller.envs" . | indent 8 }}
       containers:
-      {{- range $key := (list "low" "middle" "high") }}
-      - name: drycc-controller-celery-{{$key}}
+      - name: drycc-controller-celery
         image: {{$.Values.imageRegistry}}/{{$.Values.imageOrg}}/controller:{{$.Values.imageTag}}
         imagePullPolicy: {{$.Values.imagePullPolicy}}
         {{- if $.Values.diagnosticMode.enabled }}
@@ -50,8 +49,7 @@ spec:
         args:
           - /bin/bash
           - -c
-          - celery -A api worker -Q {{$key}} --autoscale=32,1 --loglevel=WARNING
+          - celery --app api worker --queues controller.low,controller.middle,controller.high --autoscale=32,1 --loglevel=WARNING
         {{- end }}
         {{- include "controller.limits" $ | indent 8 }}
         {{- include "controller.envs" $ | indent 8 }}
-      {{- end }}

rootfs/api/exceptions.py

@@ -39,7 +39,6 @@ def custom_exception_handler(exc, context):
     # Call REST framework's default exception handler after specific 404 handling,
     # to get the standard error response.
     response = exception_handler(exc, context)
-
     # No response means DRF couldn't handle it
     # Output a generic 500 in a JSON format
     if response is None:

rootfs/api/management/commands/cluster_lock.py

@@ -24,7 +24,7 @@ def lock(self):
         print("lock completed!")
 
     def unlock(self):
-        cache.set(lock_key, settings.VERSION)
+        cache.set(lock_key, settings.VERSION, timeout=None)
         print("unlock completed!")
 
     def waitting(self):

rootfs/api/migrations/0004_build_dryccfile_release_state_alter_service_ports.py

@@ -0,0 +1,30 @@
+# Generated by Django 4.2.10 on 2024-04-03 02:03
+
+import api.models.service
+from django.db import migrations, models
+import functools
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0003_limitspec_remove_app_procfile_structure_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='build',
+            name='dryccfile',
+            field=models.JSONField(blank=True, default=dict),
+        ),
+        migrations.AddField(
+            model_name='release',
+            name='state',
+            field=models.TextField(choices=[('created', 'Release created but not deployed'), ('crashed', 'Release pipeline runtime crashed'), ('succeed', 'Release pipeline runtime succeed')], default='created'),
+        ),
+        migrations.AlterField(
+            model_name='service',
+            name='ports',
+            field=models.JSONField(default=list, validators=[functools.partial(api.models.service.validate_json, *(), **{'schema': {'$schema': 'http://json-schema.org/schema#', 'items': {'properties': {'name': {'type': 'string'}, 'port': {'type': 'integer'}, 'protocol': {'type': 'string'}, 'targetPort': {'type': 'integer'}}, 'required': ['name', 'port', 'protocol', 'targetPort'], 'type': 'object'}, 'minItems': 1, 'type': 'array'}})]),
+        ),
+    ]

rootfs/api/models/app.py

@@ -3,6 +3,7 @@
 from django.db import models
 from django.contrib.auth import get_user_model
 from api.exceptions import DryccException, Conflict
+from api.tasks import run_pipeline
 from .base import UuidAuditedModel
 
 User = get_user_model()
@@ -22,6 +23,7 @@ class Build(UuidAuditedModel):
     # optional fields populated by builder
     sha = models.CharField(max_length=40, blank=True)
     procfile = models.JSONField(default=dict, blank=True)
+    dryccfile = models.JSONField(default=dict, blank=True)
     dockerfile = models.TextField(blank=True)
 
     class Meta:
@@ -40,6 +42,12 @@ def type(self):
             # container image (or any sort of image) used via drycc pull
             return 'image'
 
+    @property
+    def procfile_types(self):
+        if self.dryccfile:
+            return list(self.dryccfile['deploy'].keys())
+        return list(self.procfile.keys())
+
     @property
     def source_based(self):
         """
@@ -53,7 +61,16 @@ def source_based(self):
     def version(self):
         return 'git-{}'.format(self.sha) if self.source_based else 'latest'
 
-    def create(self, user, *args, **kwargs):
+    def get_image(self, procfile_type, default_image=None):
+        docker = self.dryccfile.get('build', {}).get('docker', {})
+        if procfile_type in docker:
+            if procfile_type == 'web':
+                return self.image
+            else:
+                return f'{self.image}-{procfile_type}'
+        return default_image if default_image else self.image
+
+    def create_release(self, user, *args, **kwargs):
         app_settings = self.app.appsettings_set.latest()
         latest_release = self.app.release_set.filter(failed=False).latest()
         latest_version = self.app.release_set.latest().version
@@ -64,22 +81,21 @@ def create(self, user, *args, **kwargs):
                 config=latest_release.config,
                 canary=len(app_settings.canaries) > 0,
             )
-            self.app.deploy(new_release)
+            run_pipeline.delay(new_release)
             return new_release
         except Exception as e:
             # check if the exception is during create or publish
             if ('new_release' not in locals() and
                     self.app.release_set.latest().version == latest_version+1):
                 new_release = self.app.release_set.latest()
-            if 'new_release' in locals():
+                new_release.state = "crashed"
                 new_release.failed = True
                 new_release.summary = "{} deployed {} which failed".format(self.owner, str(self.uuid)[:7])  # noqa
                 # Get the exception that has occured
                 new_release.exception = "error: {}".format(str(e))
                 new_release.save()
-            else:
+            if 'new_release' not in locals():
                 self.delete()
-
             raise DryccException(str(e)) from e
 
     def save(self, **kwargs):
@@ -90,14 +106,14 @@ def save(self, **kwargs):
             # previous release had a Procfile and the current one does not
             (
                 previous_release.build is not None and
-                len(previous_release.build.procfile) > 0 and
-                len(self.procfile) == 0
+                len(previous_release.procfile_types) > 0 and
+                len(self.procfile_types) == 0
             )
         ):
             # Reject deployment
             raise Conflict(
-                'Last deployment had a Procfile but is missing in this deploy. '
-                'For a successful deployment provide a Procfile.'
+                'Last deployment had process types but is missing in this deploy. '
+                'For a successful deployment provide process types.'
             )
 
         # See if processes are permitted to be removed
@@ -107,16 +123,16 @@ def save(self, **kwargs):
             # previous release had a Procfile and the current one does as well
             (
                 previous_release.build is not None and
-                len(previous_release.build.procfile) > 0 and
-                len(self.procfile) > 0
+                len(previous_release.procfile_types) > 0 and
+                len(self.procfile_types) > 0
             )
         )
 
         # spin down any proc type removed between the last procfile and the newest one
         if remove_procs and previous_release.build is not None:
             removed = {}
-            for proc in previous_release.build.procfile:
-                if proc not in self.procfile and self.app.structure.get(proc, 0) > 0:
+            for proc in previous_release.procfile_types:
+                if proc not in self.procfile_types and self.app.structure.get(proc, 0) > 0:
                     # Scale proc type down to 0
                     removed[proc] = 0
 
@@ -127,10 +143,11 @@ def save(self, **kwargs):
         if (
             settings.DRYCC_DEPLOY_PROCFILE_MISSING_REMOVE is False and
             previous_release.build is not None and
-            len(previous_release.build.procfile) > 0 and
-            len(self.procfile) == 0
+            len(previous_release.procfile_types) > 0 and
+            len(self.procfile_types) == 0
         ):
             self.procfile = previous_release.build.procfile
+            self.dryccfile = previous_release.build.dryccfile
 
         return super(Build, self).save(**kwargs)
 

rootfs/api/models/build.py

@@ -140,7 +140,7 @@ def _set_limits(self, previous_config):
                 if key not in data:
                     raise UnprocessableEntity(
                         '{} does not exist under {}'.format(key, 'limits'))
-                if key in self.app.types:
+                if key in self.app.procfile_types:
                     raise UnprocessableEntity(
                         "the %s has already been used and cannot be deleted" % key)
                 else:

rootfs/api/models/config.py

@@ -4,11 +4,13 @@
 from django.db import models
 from django.contrib.auth import get_user_model
 from api.utils import dict_diff
+from api.tasks import run_pipeline
 from api.exceptions import DryccException, AlreadyExists
 from scheduler import KubeHTTPException
 from scheduler.resources.pod import DEFAULT_CONTAINER_PORT
 from .base import UuidAuditedModel
 
+
 User = get_user_model()
 logger = logging.getLogger(__name__)
 
@@ -19,9 +21,14 @@ class Release(UuidAuditedModel):
 
     Releases contain a :class:`Build` and a :class:`Config`.
     """
-
+    STATE_CHOICES = (
+        ("created", "Release created but not deployed"),
+        ("crashed", "Release pipeline runtime crashed"),
+        ("succeed", "Release pipeline runtime succeed"),
+    )
     owner = models.ForeignKey(User, on_delete=models.PROTECT)
     app = models.ForeignKey('App', on_delete=models.CASCADE)
+    state = models.TextField(choices=STATE_CHOICES, default=STATE_CHOICES[0][0])
     version = models.PositiveIntegerField()
     summary = models.TextField(blank=True, null=True)
     failed = models.BooleanField(default=False)
@@ -40,8 +47,67 @@ def __str__(self):
         return "{0}-v{1}".format(self.app.id, self.version)
 
     @property
-    def image(self):
-        return self.build.image
+    def procfile_types(self):
+        if self.build is not None:
+            return self.build.procfile_types
+        return []
+
+    def get_run_image(self):
+        """
+        In the run phase of dryccfile
+        Return the kubernetes "container image" to be sent off to the scheduler.
+        """
+        image = self.build.dryccfile.get('run', {}).get(
+            'image', self.build.get_image('run'))
+        return self.build.get_image(image, default_image=image)
+
+    def get_run_args(self):
+        """
+        In the run phase of dryccfile
+        Return the kubernetes "container arguments" to be sent off to the scheduler.
+        """
+        return self.build.dryccfile.get('run', {}).get('args', [])
+
+    def get_run_command(self):
+        """
+        In the run phase of dryccfile
+        Return the kubernetes "container command" to be sent off to the scheduler.
+        """
+        return self.build.dryccfile.get('run', {}).get('command', [])
+
+    def get_deploy_image(self, container_type):
+        """
+        In the deploy phase of dryccfile
+        Return the kubernetes "container image" to be sent off to the scheduler.
+        """
+        image = self.build.dryccfile.get('deploy', {}).get(container_type, {}).get(
+            'image', self.build.get_image(container_type))
+        return self.build.get_image(image, default_image=image)
+
+    def get_deploy_args(self, container_type):
+        """
+        In the deploy phase of dryccfile
+        Return the kubernetes "container arguments" to be sent off to the scheduler.
+        """
+        if self.build is not None:
+            if self.build.dryccfile:
+                return self.build.dryccfile['deploy'].get(container_type, {}).get('args', [])
+            else:
+                # dockerfile or container image
+                if self.build.dockerfile or not self.build.sha:
+                    # has profile
+                    if self.build.procfile and container_type in self.build.procfile:
+                        args = self.build.procfile[container_type]
+                        return args.split()
+        return []
+
+    def get_deploy_command(self, container_type):
+        """
+        In the deploy phase of dryccfile
+        Return the kubernetes "container command" to be sent off to the scheduler.
+        """
+        return self.build.dryccfile.get(
+            'deploy', {}).get(container_type, {}).get('command', [])
 
     def log(self, message, level=logging.INFO):
         """Logs a message in the context of this application.
@@ -159,14 +225,14 @@ def rollback(self, user, version=None):
             )
 
             if self.build is not None:
-                self.app.deploy(new_release, force_deploy=True)
+                run_pipeline.delay(new_release, force_deploy=True)
             return new_release
         except Exception as e:
             # check if the exception is during create or publish
             if ('new_release' not in locals() and 'latest_version' in locals() and
                     self.app.release_set.latest().version == latest_version+1):
                 new_release = self.app.release_set.latest()
-            if 'new_release' in locals():
+                new_release.state = "crashed"
                 new_release.failed = True
                 new_release.summary = "{} performed roll back to a release that failed".format(self.owner)  # noqa
                 # Get the exception that has occured

rootfs/api/models/release.py

@@ -72,18 +72,21 @@ def port_name(self, port, protocol):
         return "-".join([self.app.id, self.procfile_type, protocol, str(port)]).lower()
 
     def get_port(self, port, protocol):
-        for port in self.ports:
-            if port["port"] == port and port["protocol"] == protocol:
-                return port
+        for item in self.ports:
+            if item["port"] == port and item["protocol"] == protocol:
+                return item
         return None
 
     def add_port(self, port, protocol, target_port):
-        self.ports.append({
-            "name": self.port_name(port, protocol),
-            "port": port,
-            "protocol": protocol,
-            "targetPort": target_port,
-        })
+        if self.get_port(port, protocol) is None:
+            self.ports.append({
+                "name": self.port_name(port, protocol),
+                "port": port,
+                "protocol": protocol,
+                "targetPort": target_port,
+            })
+            return True
+        return False
 
     def update_port(self, port, protocol, target_port):
         item = self.get_port(port, protocol)
@@ -167,10 +170,4 @@ def _refresh_k8s_svc(self, svc_name):
 
     def _delete_k8s_svc(self, svc_name):
         self.log('deleting Service: {}'.format(svc_name), level=logging.DEBUG)
-        try:
-            self.scheduler().svc.delete(self._namespace(), svc_name)
-        except KubeException:
-            # swallow exception
-            # raise ServiceUnavailable('Kubernetes service could not be deleted') from e
-            self.log('Kubernetes service cannot be deleted: {}'.format(svc_name),
-                     level=logging.ERROR)
+        self.scheduler().svc.delete(self._namespace(), svc_name, ignore_exception=True)

rootfs/api/models/service.py

@@ -116,7 +116,7 @@ def query_loadbalancer(namespaces: Iterator[str],
     response = requests.get(
         urljoin(settings.DRYCC_PROMETHEUS_URL, "/api/v1/query"), params=params)
     if response.status_code != 200:
-        return StopIteration
+        return
     yield from (metric["metric"] for metric in response.json()["data"]["result"])