feat(config): add typed_values

Author: duanhongyi

rootfs/api/migrations/0005_config_typed_values.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.10 on 2024-04-26 04:55
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0004_build_dryccfile_release_state_alter_service_ports'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='config',
+            name='typed_values',
+            field=models.JSONField(blank=True, default=dict),
+        ),
+    ]

rootfs/api/models/app.py

@@ -268,7 +268,7 @@ def pipeline(self, release, force_deploy=False, rollback_on_failure=True):
         release.save()
         self.log(f"{prefix} run completed...")
 
-    def deploy(self, release, structure=None, force_deploy=False, rollback_on_failure=True):
+    def deploy(self, release, procfile_types=None, force_deploy=False, rollback_on_failure=True):
         """
         Deploy a new release to this application
 
@@ -296,16 +296,17 @@ def deploy(self, release, structure=None, force_deploy=False, rollback_on_failur
         volumes = self.volume_set.all()
         deploys = {}
         for scale_type, replicas in self.structure.items():
-            if structure is not None and scale_type not in structure:
+            if procfile_types is not None and scale_type not in procfile_types:
                 continue
             scale_type_volumes = [_ for _ in volumes if scale_type in _.path.keys()]
             if not release.canary or scale_type in app_settings.canaries:
                 deploys[scale_type] = self._gather_app_settings(
                     release, app_settings, scale_type, replicas, volumes=scale_type_volumes)
-        self._deploy(deploys, structure, prev_release, release, force_deploy, rollback_on_failure)
+        self._deploy(
+            deploys, procfile_types, prev_release, release, force_deploy, rollback_on_failure)
         # cleanup old release objects from kubernetes
-        self.cleanup_old()
-        release.cleanup_old()
+        self.cleanup_old(procfile_types)
+        release.cleanup_old(procfile_types)
 
     def mount(self, user, volume, structure=None):
         if self.release_set.filter(failed=False).latest().build is None:
@@ -320,15 +321,17 @@ def mount(self, user, volume, structure=None):
                 app_settings,
                 structure=structure,
             )
-        self._mount(user, volume, release, app_settings)
+        self._mount(user, volume, release, app_settings, structure=structure)
 
-    def cleanup_old(self):
+    def cleanup_old(self, procfile_types=None):
         names, app_settings = [], self.appsettings_set.latest()
         for scale_type in self.structure.keys():
             if scale_type in app_settings.canaries:
                 names.append(self._get_job_id(scale_type, True))
             names.append(self._get_job_id(scale_type, False))
         labels = {'heritage': 'drycc'}
+        if procfile_types:
+            labels["type__in"] = procfile_types
         deployments = self.scheduler().deployments.get(self.id, labels=labels).json()["items"]
         if deployments is not None:
             for deployment in deployments:
@@ -390,12 +393,12 @@ def pod_name(size=5, chars=string.ascii_lowercase + string.digits):
         data['restart_policy'] = 'Never'
         data['active_deadline_seconds'] = timeout
         data['ttl_seconds_after_finished'] = expires
-        # create application config and build the pod manifest
-        self.set_application_config(release)
         name = self._get_job_id(PROCFILE_TYPE_RUN, release.canary) + '-' + pod_name()
         self.log("{} on {} runs '{}'".format(user.username, name, command))
         kwargs.update(data)
         try:
+            # create application config and build the pod manifest
+            self.set_application_config(release, PROCFILE_TYPE_RUN)
             self.scheduler().job.create(
                 self.id,
                 name,
@@ -506,7 +509,7 @@ def image_pull_secret(self, namespace, registry, image):
 
         return name
 
-    def set_application_config(self, release):
+    def set_application_config(self, release, procfile_type):
         """
         Creates the application config as a secret in Kubernetes and
         updates it if it already exists
@@ -515,18 +518,19 @@ def set_application_config(self, release):
         version = 'v{}'.format(release.version)
         labels = {
             'version': version,
-            'type': 'env'
+            'type': procfile_type,
+            'class': 'env'
         }
 
         # secrets use dns labels for keys, map those properly here
         secrets_env = {}
-        for key, value in self._build_env_vars(release).items():
+        for key, value in self._build_env_vars(release, procfile_type).items():
             secrets_env[key.lower().replace('_', '-')] = str(value)
 
         # dictionary sorted by key
         secrets_env = OrderedDict(sorted(secrets_env.items(), key=lambda t: t[0]))
 
-        secret_name = "{}-{}-env".format(self.id, version)
+        secret_name = "{}-{}-{}-env".format(self.id, procfile_type, version)
         try:
             self.scheduler().secret.get(self.id, secret_name)
         except KubeHTTPException:
@@ -578,7 +582,8 @@ def _mount(self, user, volume, release, app_settings, structure=None):
         volumes = Volume.objects.filter(app=self)
         tasks = []
         for scale_type, replicas in structure.items() if structure else self.structure.items():
-            if not release.canary or scale_type in app_settings.canaries:
+            if scale_type != PROCFILE_TYPE_RUN and (
+                    not release.canary or scale_type in app_settings.canaries):
                 replicas = self.structure.get(scale_type, 0)
                 scale_type_volumes = [
                     volume for volume in volumes if scale_type in volume.path.keys()]
@@ -588,53 +593,47 @@ def _mount(self, user, volume, release, app_settings, structure=None):
                     self.id, self._get_job_id(scale_type, release.canary)).json()
                 spec_annotations = deployment['spec']['template']['metadata'].get(
                     'annotations', {})
+                self.set_application_config(release, scale_type)
                 # gather volume proc types to be deployed
-                tasks.append(
-                    functools.partial(
-                        self.scheduler().deployment.patch,
-                        namespace=self.id,
-                        name=self._get_job_id(scale_type, release.canary),
-                        image=release.get_deploy_image(scale_type),
-                        command=release.get_deploy_command(scale_type),
-                        args=release.get_deploy_args(scale_type),
-                        spec_annotations=spec_annotations,
-                        resource_version=deployment["metadata"]["resourceVersion"],
-                        **data
-                    )
-                )
+                tasks.append(functools.partial(
+                    self.scheduler().deployment.patch,
+                    namespace=self.id,
+                    name=self._get_job_id(scale_type, release.canary),
+                    image=release.get_deploy_image(scale_type),
+                    command=release.get_deploy_command(scale_type),
+                    args=release.get_deploy_args(scale_type),
+                    spec_annotations=spec_annotations,
+                    resource_version=deployment["metadata"]["resourceVersion"],
+                    **data
+                ))
         try:
-            # create the application config in k8s (secret in this case) for all deploy objects
-            self.set_application_config(release)
             apply_tasks(tasks)
         except Exception as e:
             err = f'(changed volume mount for {volume}: {e}'
             self.log(err, logging.ERROR)
             raise ServiceUnavailable(err) from e
         self.log(f'{user.username} changed volume mount for {volume}')
 
-    def _deploy(self, deploys, structure, prev_release,
+    def _deploy(self, deploys, procfile_types, prev_release,
                 release, force_deploy, rollback_on_failure):
         # Sort deploys so routable comes first
         deploys = OrderedDict(sorted(deploys.items(), key=lambda d: d[1].get('routable')))
         # Check if any proc type has a Deployment in progress
         self._check_deployment_in_progress(deploys, release, force_deploy)
 
         try:
-            # create the application config in k8s (secret in this case) for all deploy objects
-            self.set_application_config(release)
-
-            # gather all proc types to be deployed
-            tasks = [
-                functools.partial(
+            tasks = []
+            for scale_type, kwargs in deploys.items():
+                self.set_application_config(release, scale_type)
+                tasks.append(functools.partial(
                     self.scheduler().deploy,
                     namespace=self.id,
                     name=self._get_job_id(scale_type, release.canary),
                     image=release.get_deploy_image(scale_type),
                     command=release.get_deploy_command(scale_type),
                     args=release.get_deploy_args(scale_type),
                     **kwargs
-                ) for scale_type, kwargs in deploys.items()
-            ]
+                ))
             try:
                 apply_tasks(tasks)
             except KubeException as e:
@@ -647,7 +646,7 @@ def _deploy(self, deploys, structure, prev_release,
                     # This goes in the log before the rollback starts
                     self.log(err, logging.ERROR)
                     # revert all process types to old release
-                    self.deploy(prev_release, structure,
+                    self.deploy(prev_release, procfile_types,
                                 force_deploy=True, rollback_on_failure=False)
                     # let it bubble up
                     raise DryccException('{}\n{}'.format(err, str(e))) from e
@@ -721,6 +720,8 @@ def _scale_pods(self, scale_types, release, app_settings):
                 volume for volume in volumes if scale_type in volume.path.keys()]
             data = self._gather_app_settings(
                 release, app_settings, scale_type, replicas, volumes=scale_type_volumes)
+            # create the application config in k8s (secret in this case) for all deploy objects
+            self.set_application_config(release, scale_type)
             # gather all proc types to be deployed
             tasks.append(
                 functools.partial(
@@ -734,8 +735,6 @@ def _scale_pods(self, scale_types, release, app_settings):
                 )
             )
         try:
-            # create the application config in k8s (secret in this case) for all deploy objects
-            self.set_application_config(release)
             apply_tasks(tasks)
         except Exception as e:
             err = '(scale): {}'.format(e)
@@ -912,7 +911,7 @@ def _scheduler_filter(self, **kwargs):
             labels.update({'version': version})
         return labels
 
-    def _build_env_vars(self, release):
+    def _build_env_vars(self, release, procfile_type):
         """
         Build a dict of env vars, setting default vars based on app type
         and then combining with the user set ones
@@ -935,9 +934,9 @@ def _build_env_vars(self, release):
         port = release.get_port()
         if port:
             default_env['PORT'] = port
-
         # merge envs on top of default to make envs win
         default_env.update(release.config.values)
+        default_env.update(release.config.typed_values.get(procfile_type, {}))
         return default_env
 
     def _get_private_registry_config(self, image, registry=None):
@@ -999,7 +998,7 @@ def _gather_app_settings(self, release, app_settings, procfile_type, replicas, v
         Any global setting that can also be set per app goes here
         """
 
-        envs = self._build_env_vars(release)
+        envs = self._build_env_vars(release, procfile_type)
         # Obtain a limit plan that must exist, if raise error here, it must be a bug
         config = self._set_default_limit(release.config, procfile_type)
         limit_plan = LimitPlan.objects.get(id=config.limits.get(procfile_type))

rootfs/api/models/config.py

@@ -18,12 +18,13 @@ class Config(UuidAuditedModel):
     during runtime execution of the Application.
     """
     procfile_fields = ("lifecycle_post_start", "lifecycle_pre_stop", "tags", "limits",
-                       "healthcheck", "termination_grace_period")
+                       "typed_values", "healthcheck", "termination_grace_period")
     all_diff_fields = ("values", "registry") + procfile_fields
 
     owner = models.ForeignKey(User, on_delete=models.PROTECT)
     app = models.ForeignKey('App', on_delete=models.CASCADE)
     values = models.JSONField(default=dict, blank=True)
+    typed_values = models.JSONField(default=dict, blank=True)
     lifecycle_post_start = models.JSONField(default=dict, blank=True)
     lifecycle_pre_stop = models.JSONField(default=dict, blank=True)
     tags = models.JSONField(default=dict, blank=True)
@@ -79,17 +80,9 @@ def save(self, **kwargs):
                          'lifecycle_pre_stop', 'termination_grace_period']:
                 data = getattr(previous_config, attr, {}).copy()
                 new_data = getattr(self, attr, {}).copy()
-
-                # remove config keys if a null value is provided
-                for key, value in new_data.items():
-                    if value is None:
-                        # error if unsetting non-existing key
-                        if key not in data:
-                            raise UnprocessableEntity('{} does not exist under {}'.format(key, attr))  # noqa
-                        data.pop(key)
-                    else:
-                        data[key] = value
+                self._merge_data(attr, data, new_data)
                 setattr(self, attr, data)
+            self._set_typed_values(previous_config)
             self._set_limits(previous_config)
             self._set_healthcheck(previous_config)
             self._set_registry()
@@ -99,6 +92,18 @@ def save(self, **kwargs):
 
         return super(Config, self).save(**kwargs)
 
+    def _merge_data(self, field, data, new_data):
+        # remove config keys if a null value is provided
+        for key, value in new_data.items():
+            if value is None:
+                # error if unsetting non-existing key
+                if key not in data:
+                    raise UnprocessableEntity('{} does not exist under {}'.format(key, field))  # noqa
+                data.pop(key)
+            else:
+                data[key] = value
+        return data
+
     def _set_registry(self):
         # lower case all registry options for consistency
         self.registry = {key.lower(): value for key, value in self.registry.copy().items()}
@@ -144,20 +149,13 @@ def _set_tags(self, previous_config):
     def _set_limits(self, previous_config):
         data = getattr(previous_config, 'limits', {}).copy()
         new_data = getattr(self, 'limits', {}).copy()
-        # remove config keys if a null value is provided
+        # check procfile
         for key, value in new_data.items():
             if value is None:
-                # error if unsetting non-existing key
-                if key not in data:
-                    raise UnprocessableEntity(
-                        '{} does not exist under {}'.format(key, 'limits'))
                 if key in self.app.procfile_types:
                     raise UnprocessableEntity(
                         "the %s has already been used and cannot be deleted" % key)
-                else:
-                    data.pop(key)
-            else:
-                data[key] = value
+        self._merge_data('limits', data, new_data)
         setattr(self, 'limits', data)
 
     def _set_healthcheck(self, previous_config):
@@ -168,17 +166,25 @@ def _set_healthcheck(self, previous_config):
             if value is None:
                 # error if unsetting non-existing key
                 if key not in data:
-                    raise UnprocessableEntity('{} does not exist under {}'.format(key, 'healthcheck'))  # noqa
+                    raise UnprocessableEntity(
+                        '{} does not exist under {}'.format(key, 'healthcheck'))
                 data.pop(key)
             else:
-                for probeType, probe in value.items():
-                    if probe is None:
-                        # error if unsetting non-existing key
-                        if key not in data or probeType not in data[key].keys():
-                            raise UnprocessableEntity('{} does not exist under {}'.format(key, 'healthcheck'))  # noqa
-                        data[key].pop(probeType)
-                    else:
-                        if key not in data:
-                            data[key] = {}
-                        data[key][probeType] = probe
+                data[key] = self._merge_data('healthcheck', data.get(key, {}), value)
         setattr(self, 'healthcheck', data)
+
+    def _set_typed_values(self, previous_config):
+        data = getattr(previous_config, 'typed_values', {}).copy()
+        new_data = getattr(self, 'typed_values', {}).copy()
+        # remove config keys if a null value is provided
+        for procfile_type, values in new_data.items():
+            if not values:
+                # error if unsetting non-existing key
+                if procfile_type not in data:
+                    raise UnprocessableEntity(
+                        '{} does not exist under {}'.format(procfile_type, 'typed_values'))
+                data.pop(procfile_type)
+            else:
+                data[procfile_type] = self._merge_data(
+                    'typed_values', data.get(procfile_type, {}), values)
+        setattr(self, 'typed_values', data)