feat(dg-auto-update): dg auto update distributor and action

nuvla/job_engine/job/actions/deployment_set_automatic_update.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+
+from ..actions import action
+
+import logging
+
+@action('deployment_set_automatic_update')
+class DeploymentSetAutomaticUpdateJob(object):
+
+    def __init__(self, job):
+        self.job = job
+        self.api = job.api
+        self.dep_set_id = self.job['target-resource']['href']
+
+    def _auto_update(self):
+        try:
+            dep_set = self.api.get(self.dep_set_id)
+            operations = map(lambda x: x['rel'], dep_set.data['operations'])
+            if 'auto-update' in operations:
+                self.api.operation(dep_set, 'auto-update')
+                logging.info(f'Deployment set auto updated: {self.dep_set_id}')
+        except Exception as ex:
+            logging.error(f'Failed to auto update {self.dep_set_id}: {repr(ex)}')
+
+    def do_work(self):
+        logging.info(f'Deployment set automatic update {self.job.id}')
+        self._auto_update()
+        logging.info(f'End of deployment set automatic update {self.job.id}')
+        return 0

nuvla/job_engine/job/distributions/deployment_set_automatic_update.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+
+import logging
+
+from nuvla.api.util.filter import filter_and
+from ..job import JOB_QUEUED, JOB_RUNNING
+from ..util import override
+from ..distributions import distribution
+from ..distribution import DistributionBase
+
+
+@distribution('deployment_set_automatic_update')
+class DeploymentSetAutomaticUpdateJobsDistribution(DistributionBase):
+    DISTRIBUTION_NAME = 'deployment_set_automatic_update'
+
+    def __init__(self, distributor):
+        super(DeploymentSetAutomaticUpdateJobsDistribution, self).__init__(
+            self.DISTRIBUTION_NAME, distributor)
+        self.collect_interval = 60
+        self._start_distribution()
+
+    def auto_update_dgs(self):
+        try:
+            return self.distributor.api.search(
+                'deployment-set',
+                filter=filter_and(['state=["STARTED","UPDATED","PARTIALLY-STARTED","PARTIALLY-UPDATED"]',
+                                   'auto-update=true',
+                                   'next-refresh<="now"']),
+                select='id',
+                last=10000).resources
+        except Exception as ex:
+            logging.error(f'Failed to search for auto-update dgs: {ex}')
+            return []
+
+    def job_exists(self, job):
+        jobs = self.distributor.api.search(
+            'job',
+            filter=filter_and(
+                [f'state={[JOB_RUNNING, JOB_QUEUED]}',
+                 f"action='{job['action']}'",
+                 f"target-resource/href='{job['target-resource']['href']}'"]),
+            last=0)
+        return jobs.count > 0
+
+    @override
+    def job_generator(self):
+        auto_update_dgs = self.auto_update_dgs()
+        logging.info(f'Auto-update DGs count: {len(auto_update_dgs)}')
+        for resource in auto_update_dgs:
+            job = {'action': self.DISTRIBUTION_NAME,
+                   'target-resource': {'href': resource.id}}
+            if self.job_exists(job):
+                continue
+            yield job