feat(deploy): finer grained ptype locks

Author: duanhongyi

rootfs/api/models/app.py

@@ -249,10 +249,14 @@ def restart(self, **kwargs):  # noqa
             deployments.append(self._get_job_id(kwargs['type']))
         try:
             tasks = [
-                functools.partial(
-                    self.scheduler().deployment.restart,
-                    self.id,
-                    deployment
+                (
+                    functools.partial(
+                        self.scheduler().deployment.restart,
+                        self.id,
+                        deployment
+                    ),
+                    lambda future: self.log(
+                        f'restart {kwargs['type']} callback: {future.result()}'),
                 ) for deployment in deployments
             ]
             apply_tasks(tasks)
@@ -273,7 +277,7 @@ def scale(self, user, structure):
         app_settings = self.appsettings_set.latest()
         self._scale(user, structure, release, app_settings)
 
-    def pipeline(self, release, ptypes=None, force_deploy=False):
+    def pipeline(self, release, ptypes, force_deploy=False):
         prefix = f"[pipeline] release {release.version_name}"
         try:
             if release.build is not None:
@@ -306,7 +310,7 @@ def pipeline(self, release, ptypes=None, force_deploy=False):
                 state="crashed", action="pipeline", ptypes=ptypes, exception=str(e))
             self.log(f"{prefix} pipeline runtime error: {release.exception}", logging.ERROR)
         finally:
-            DeployLock(self.pk).release(ptypes)
+            DeployLock(self.pk).release(ptypes)  # release all locks
             release.save(update_fields=["state", "failed"])  # avoid overwriting other fields
         self.log(f"{prefix} run completed...")
 
@@ -744,16 +748,20 @@ def _mount(self, user, volume, release, app_settings, structure=None):
                     '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),
-                    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),
+                        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
+                    ),
+                    lambda future: self.log(
+                        f'mount {volume} for {scale_type} callback: {future.result()}'),
                 ))
         try:
             apply_tasks(tasks)
@@ -769,19 +777,24 @@ def _deploy(self, deploys, ptypes, prev_release,
         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, force_deploy)
-
         try:
             tasks = []
+            lock = DeployLock(self.pk)
             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),
-                    image=release.get_deploy_image(scale_type),
-                    command=release.get_deploy_command(scale_type),
-                    args=release.get_deploy_args(scale_type),
-                    **kwargs
+                tasks.append((
+                    functools.partial(
+                        self.scheduler().deploy,
+                        namespace=self.id,
+                        name=self._get_job_id(scale_type),
+                        image=release.get_deploy_image(scale_type),
+                        command=release.get_deploy_command(scale_type),
+                        args=release.get_deploy_args(scale_type),
+                        **kwargs
+                    ),
+                    lambda future: self.log(
+                        f'deploy and unlock callback: {[
+                            future.result(), lock.release([scale_type])]}'),
                 ))
             try:
                 apply_tasks(tasks)
@@ -797,7 +810,6 @@ def _deploy(self, deploys, ptypes, prev_release,
                     self.deploy(prev_release, ptypes, True, False)
                     # let it bubble up
                     raise DryccException('{}\n{}'.format(err, str(e))) from e
-
                 # otherwise just re-raise
                 raise
         except Exception as e:
@@ -876,7 +888,7 @@ def _scale_pods(self, scale_types, release, app_settings):
             # 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(
+            tasks.append((
                 functools.partial(
                     self.scheduler().scale,
                     namespace=self.id,
@@ -885,8 +897,9 @@ def _scale_pods(self, scale_types, release, app_settings):
                     command=release.get_deploy_command(scale_type),
                     args=release.get_deploy_args(scale_type),
                     **data
-                )
-            )
+                ),
+                lambda future: self.log(f'scale {scale_type} callback: {future.result()}'),
+            ))
         try:
             apply_tasks(tasks)
         except Exception as e:
@@ -1046,10 +1059,9 @@ def _check_deployment_in_progress(self, deploys, force_deploy=False):
     @staticmethod
     def _default_structure(release):
         """Scale to default structure based on release type"""
-        structure = {PTYPE_WEB: 1}
+        structure = {}
         for ptype in release.ptypes:
-            if ptype != PTYPE_WEB:
-                structure[ptype] = 0
+            structure[ptype] = 1 if ptype == PTYPE_WEB else 0
         return structure
 
     def _scheduler_filter(self, **kwargs):

rootfs/api/models/build.py

@@ -3,7 +3,7 @@
 from django.db import models
 from django.contrib.auth import get_user_model
 from api.exceptions import DryccException, Conflict
-from .base import UuidAuditedModel
+from .base import UuidAuditedModel, PTYPE_WEB
 
 User = get_user_model()
 logger = logging.getLogger(__name__)
@@ -45,7 +45,9 @@ def type(self):
     def ptypes(self):
         if self.dryccfile:
             return list(self.dryccfile['deploy'].keys())
-        return list(self.procfile.keys())
+        if self.procfile:
+            return list(self.procfile.keys())
+        return [PTYPE_WEB]
 
     @property
     def source_based(self):

rootfs/api/models/release.py

@@ -164,23 +164,26 @@ def get_port(self, ptype):
                 'PORT', self.config.values.get('PORT', DEFAULT_CONTAINER_PORT)))
 
     def deploy(self, ptypes=None, force_deploy=False):
-        msg = 'there is an executing pipeline, please wait or force deploy'
+        ptypes = set(ptypes).intersection(self.ptypes) if ptypes else self.ptypes
+        if not ptypes:
+            self.log(f'skip deploy, ptypes are not within the optional range: {self.ptypes}')
+            return
         # change deployed_ptypes lock
+        msg = 'there is an executing pipeline, please wait or force deploy'
         lock = CacheLock("release:%s" % self.pk)
         try:
             if lock.acquire() or force_deploy:
-                if ptypes is None:
-                    deployed_ptypes = self.ptypes
-                else:
-                    deployed_ptypes = list(set(self.deployed_ptypes).union(ptypes))
-                type(self).objects.filter(pk=self.pk).update(deployed_ptypes=deployed_ptypes)
+                deployed_ptypes = list(set(self.deployed_ptypes).union(ptypes))
+                if deployed_ptypes:
+                    type(self).objects.filter(pk=self.pk).update(deployed_ptypes=deployed_ptypes)
             else:
                 raise DryccException(msg)
         finally:
             lock.release()
         # deploy lock
-        if not DeployLock(self.app.pk).acquire(ptypes, force=force_deploy):
-            raise DryccException(msg)
+        lock = DeployLock(self.app.pk)
+        if not lock.acquire(ptypes, force=force_deploy):
+            raise DryccException(f"{msg}: {lock.locked(ptypes)}")
         run_pipeline.delay(self, ptypes, force_deploy)
 
     def previous(self):

rootfs/api/tests/test_build.py

@@ -172,7 +172,7 @@ def test_build_default_containers(self, mock_requests):
         url = f"/v2/apps/{app_id}/pods/"
         response = self.client.get(url)
         self.assertEqual(response.status_code, 200, response.data)
-        self.assertPodContains(response.data['results'], app_id, 'web', "v2", "up")
+        self.assertEqual(response.data['results'], [])
         # start with a new app
         app_id = self.create_app()
         # post a new build with procfile
@@ -263,13 +263,7 @@ def test_build_forgotten_procfile(self, mock_requests):
         url = f"/v2/apps/{app_id}/pods/"
         response = self.client.get(url)
         self.assertEqual(response.status_code, 200, response.data)
-        self.assertEqual(len(response.data['results']), 1)
-
-        # verify web is in there
-        url = f"/v2/apps/{app_id}/pods/"
-        response = self.client.get(url)
-        self.assertEqual(response.status_code, 200, response.data)
-        self.assertEqual(len(response.data['results']), 1)
+        self.assertEqual([item for item in response.data['results'] if item["type"] != "web"], [])
 
         # look at the app structure
         url = f"/v2/apps/{app_id}"
@@ -319,35 +313,18 @@ def test_build_no_remove_process(self, mock_requests):
         response = self.client.post(url, body)
         self.assertEqual(response.status_code, 201, response.data)
 
-        # verify worker is still there
-        url = f"/v2/apps/{app_id}/pods/"
-        response = self.client.get(url)
-        self.assertEqual(response.status_code, 200, response.data)
-        self.assertPodContains(response.data['results'], app_id, 'worker', "v3", "up")
-
-        # verify web is still there
+        # verify web
         url = f"/v2/apps/{app_id}/pods/"
         response = self.client.get(url)
         self.assertEqual(response.status_code, 200, response.data)
+        self.assertEqual([item for item in response.data['results'] if item["type"] != "web"], [])
         self.assertPodContains(response.data['results'], app_id, 'web', "v3", "up")
 
         # look at the app structure
         url = f"/v2/apps/{app_id}"
         response = self.client.get(url)
         self.assertEqual(response.status_code, 200, response.data)
-        self.assertEqual(response.json()['structure'], {'web': 1, 'worker': 1})
-
-        # scale worker to make sure no info was lost
-        url = f"/v2/apps/{app_id}/ptypes/scale"
-        body = {'worker': 2}  # bump from 1 to 2
-        response = self.client.post(url, body)
-        self.assertEqual(response.status_code, 204, response.data)
-
-        # verify worker info
-        url = f"/v2/apps/{app_id}/pods/"
-        response = self.client.get(url)
-        self.assertEqual(response.status_code, 200, response.data)
-        self.assertPodContains(response.data['results'], app_id, 'worker', "v3", "up")
+        self.assertEqual(response.json()['structure'], {'web': 1})
 
     @override_settings(DRYCC_DEPLOY_REJECT_IF_PROCFILE_MISSING=True)
     def test_build_forgotten_procfile_reject(self, mock_requests):
@@ -389,7 +366,12 @@ def test_build_forgotten_procfile_reject(self, mock_requests):
         url = f"/v2/apps/{app_id}/build"
         body = {'image': 'autotest/example', 'stack': 'container'}
         response = self.client.post(url, body)
-        self.assertEqual(response.status_code, 409, response.data)
+        self.assertEqual(response.status_code, 201, response.data)
+        # verify web
+        url = f"/v2/apps/{app_id}/pods/"
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200, response.data)
+        self.assertPodContains(response.data['results'], app_id, 'web', "v3", "up")
 
     def test_build_str(self, mock_requests):
         """Test the text representation of a build."""

rootfs/api/tests/test_lock.py

@@ -25,24 +25,19 @@ def test_deploy_lock(self):
         app_id = f"test_key_1_{int(time.time())}"
         lock1 = DeployLock(app_id)
         lock2 = DeployLock(app_id)
-        self.assertEqual(lock1.acquire(None), True)
-        self.assertEqual(lock2.acquire(["web", "task"]), False)
-        lock1.release(None)
         self.assertEqual(lock2.acquire(["web", "task"]), True)
-        self.assertEqual(lock1.acquire(None), False)
         self.assertEqual(lock1.acquire(["web"]), False)
         self.assertEqual(lock1.acquire(["bing"]), True)
         lock2.release(["web", "task"])
         self.assertEqual(lock1.acquire(["web"]), True)
         self.assertEqual(lock1.acquire(["task"]), True)
         self.assertEqual(lock2.acquire(["web", "task"]), False)
-        self.assertEqual(lock2.acquire(None), False)
         lock2.release(["web", "task", "bing"])
-        self.assertEqual(lock2.acquire(None), True)
-        lock2.release(None)
         self.assertEqual(lock1.acquire(["web", "task"]), True)
         self.assertEqual(lock2.acquire(["web", "task"]), False)
         self.assertEqual(lock2.acquire(["web", "bing"], True), True)
         self.assertEqual(lock1.acquire(["web", "bing", "task"]), False)
         self.assertEqual(lock1.acquire(["web", "bing", "task"], True), True)
         lock2.release(["web", "bing", "task"])
+        self.assertEqual(lock1.acquire(["web"], True), True)
+        lock1.release(["web", "bing", "task"])

rootfs/api/tests/test_pods.py

@@ -752,10 +752,12 @@ def test_modified_procfile_from_build_removes_pods(self, mock_requests):
         }
         response = self.client.post(build_url, body)
         self.assertEqual(response.status_code, 201, response.data)
-        # web is default ptype
+        # check web pods
         application = App.objects.get(id=app_id)
         pods = application.list_pods(type='web')
-        self.assertEqual(len(pods), 4)
+        self.assertEqual(len(pods), 0)
+        pods = application.list_pods(type='worker')
+        self.assertEqual(len(pods), 0)
 
     def test_list_pods_failure(self, mock_requests):
         """

rootfs/api/utils.py

@@ -160,7 +160,8 @@ def apply_tasks(tasks):
         return
 
     executor = concurrent.futures.ThreadPoolExecutor(5)
-    for future in [executor.submit(task) for task in tasks]:
+    for future, callback in [(executor.submit(task[0]), task[1]) for task in tasks]:
+        future.add_done_callback(callback)
         error = future.exception()
         if error is not None:
             raise error
@@ -231,32 +232,20 @@ def __init__(self, app_id):
 
     def locked(self, ptypes):
         value = cache.get(self.ptypes_key, [])
-        # None will locks all
-        if ptypes is None:
-            if value is None or len(value) > 0:
-                return True
-            elif len(value) == 0:
-                return False
-        else:
-            if value is None:
-                return True
-            for ptype in ptypes:
-                if ptype in value:
-                    return True
-        return False
+        locked_ptypes = []
+        for ptype in ptypes:
+            if ptype in value:
+                locked_ptypes.append(ptype)
+        return locked_ptypes
 
     def acquire(self, ptypes, force=False):
         try:
             if super(DeployLock, self).acquire():
                 value = cache.get(self.ptypes_key, [])
-                # None will locks all
-                if not force and self.locked(ptypes):
+                if not force and len(self.locked(ptypes)) > 0:
                     return False
-                if ptypes is None:
-                    value = None
-                else:
-                    value.extend(ptypes)
-                    value = list(set(value))
+                value.extend(ptypes)
+                value = list(set(value))
                 cache.set(self.ptypes_key, value, timeout=3600)
                 return True
         finally:
@@ -266,15 +255,13 @@ def acquire(self, ptypes, force=False):
     def release(self, ptypes):
         try:
             if super(DeployLock, self).acquire():
-                if ptypes is None:
-                    cache.delete(self.ptypes_key)
-                else:
-                    value = cache.get(self.ptypes_key, [])
-                    if value is None:
-                        return
-                    for ptype in ptypes:
+                value = cache.get(self.ptypes_key, [])
+                if value is None:
+                    return
+                for ptype in ptypes:
+                    if ptype in value:
                         value.remove(ptype)
-                    cache.set(self.ptypes_key, value)
+                cache.set(self.ptypes_key, value)
         finally:
             super(DeployLock, self).release()