Merge pull request #799 from MetaCell/feature/CH-55-61

CH-55 CH-61 add retry policy and customization to tasks

Author: filippomc

applications/workflows/server/workflows_api/service/workflow_service.py

@@ -1,12 +1,12 @@
-from cloudharness.workflows import argo
+from cloudharness.workflows import argo_service
 from workflows_api.models import OperationSearchResult, Operation, SearchResultData
 
-OperationNotFound = argo.WorkflowNotFound
-OperationException = argo.WorkflowException
-BadParam = argo.BadParam
+OperationNotFound = argo_service.WorkflowNotFound
+OperationException = argo_service.WorkflowException
+BadParam = argo_service.BadParam
 
 
-def argo_workflow_to_operation(workflow: argo.Workflow):
+def argo_workflow_to_operation(workflow: argo_service.Workflow):
     return Operation(name=workflow.name,
                      status=workflow.status,
                      create_time=workflow.create_time,
@@ -15,12 +15,12 @@ def argo_workflow_to_operation(workflow: argo.Workflow):
 
 def delete_operation(name):
     """deletes operation by id"""
-    argo.delete_workflow(name)
+    argo_service.delete_workflow(name)
 
 
 def get_operation(name):
     """get operation by id"""
-    return argo_workflow_to_operation(argo.get_workflow(name))
+    return argo_workflow_to_operation(argo_service.get_workflow(name))
 
 
 def list_operations(status=None, continue_token=None, limit=None) -> OperationSearchResult:
@@ -38,7 +38,7 @@ def list_operations(status=None, continue_token=None, limit=None) -> OperationSe
     :rtype: OperationSearchResult
     """
 
-    argo_raw_result = argo.get_workflows(status, limit=limit, continue_token=continue_token)
+    argo_raw_result = argo_service.get_workflows(status, limit=limit, continue_token=continue_token)
     result = OperationSearchResult()
     result.items = tuple(argo_workflow_to_operation(item) for item in argo_raw_result.items)
     result.meta = SearchResultData(continue_token=argo_raw_result.continue_token)
@@ -51,4 +51,4 @@ def log_operation(name: str) -> str:
     :rtype: str
     """
 
-    return argo.get_workflow_logs(name)
+    return argo_service.get_workflow_logs(name)

docs/applications/development/workflows-api.md

@@ -219,6 +219,18 @@ ttl_strategy={
 op = operations.ParallelOperation(..., ttl_strategy=ttl_strategy)
 ```
 
+## Retry strategy
+
+By default, workflow tasks are retried 10 times. To set the value to something else, use 
+the `retry_limit` parameter.
+
+```python
+from cloudharness.workflows import operations, tasks
+
+my_task = tasks.CustomTask('my-task-retry', 'myapp-mytask', retry_limit=2)
+op = operations.SingleTaskOperation('my-op-retry-', my_task)
+```
+
 ## Notify on exit
 
 The parameter `on_exit_notify` adds an additional task to the workflow that notifies its completion in the events queue.
@@ -235,6 +247,32 @@ op = operations.ParallelOperation(..., on_exit_notify=on_exit_notify)
 
 Synchronous operation types use this mechanism to wait for the result and get the value.
 
+To customize the onExit strategy an additional `image` parameter can be specified. 
+
+```Python
+import json
+on_exit_notify={
+    'queue': 'my_queue',
+    'payload': json.dumps({'insert': 1})
+    'image': "my-image"
+}
+op = operations.ParallelOperation(..., on_exit_notify=on_exit_notify)
+```
+
+## Free template customization
+
+To customize the task beyond the Task api, use 
+the `template_overrides` parameter.
+
+```python
+from cloudharness.workflows import operations, tasks
+
+my_task = tasks.CustomTask('my-task-retry', 'myapp-mytask', retry_limit=2)
+op = operations.SingleTaskOperation('my-op-retry-', template_overrides=tasks.V1alpha1Template(
+            memoize=True,
+        ))
+```
+
 ## Workflows query service api
 
 Workflows can be queried and retrieved through the Python api

libraries/cloudharness-common/cloudharness/utils/__init__.py

@@ -1,5 +1,7 @@
+import collections
 
-def dict_merge(dct, merge_dct, add_keys=True):
+
+def dict_merge(dct, merge_dct, add_keys=True, merge_none=True):
     """ Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
     updating only top-level keys, dict_merge recurses down into dicts nested
     to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
@@ -31,7 +33,7 @@ def dict_merge(dct, merge_dct, add_keys=True):
         if (k in dct and isinstance(dct[k], dict) and
                 isinstance(merge_dct[k], collections.abc.Mapping)):
             dct[k] = dict_merge(dct[k], merge_dct[k], add_keys=add_keys)
-        else:
+        elif merge_none or (merge_dct[k] is not None):
             dct[k] = merge_dct[k]
 
     return dct

libraries/cloudharness-common/cloudharness/workflows/__init__.py

@@ -0,0 +1,2 @@
+from . import argo_service as argo
+from . import argo_service

…ss-common/cloudharness/workflows/argo.py …n/cloudharness/workflows/argo_service.pylibraries/cloudharness-common/cloudharness/workflows/argo.py renamed to libraries/cloudharness-common/cloudharness/workflows/argo_service.py libraries/cloudharness-common/cloudharness/workflows/argo.py renamed to libraries/cloudharness-common/cloudharness/workflows/argo_service.py

@@ -15,7 +15,11 @@
 from cloudharness import log, applications
 
 ch_conf = conf.get_configuration()
-namespace = conf.get_namespace()
+try:
+    namespace = conf.get_namespace()
+except Exception as e:
+    log.error("Error getting namespace: %s", e)
+    namespace = 'default'
 
 
 class WorkflowException(Exception):

libraries/cloudharness-common/cloudharness/workflows/operations.py

@@ -9,7 +9,7 @@
 from cloudharness import log
 from cloudharness.events.client import EventClient
 from cloudharness.utils import env, config
-from . import argo
+from . import argo_service
 from .tasks import Task, SendResultTask, CustomTask
 from .utils import PodExecutionContext, affinity_spec, is_accounts_present, name_from_path, volume_mount_template
 from argo.workflows.client import V1Toleration
@@ -60,7 +60,7 @@ class ContainerizedOperation(ManagedOperation):
     """
 
     def __init__(self, basename: str, pod_context: Union[PodExecutionContext, list, tuple] = None,
-                 shared_directory=None, *args, **kwargs):
+                 shared_directory=None, pod_gc=None, *args, **kwargs):
         """
         :param basename:
         :param pod_context: PodExecutionContext - represents affinity with other pods in the system
@@ -76,6 +76,7 @@ def __init__(self, basename: str, pod_context: Union[PodExecutionContext, list,
 
         self.persisted = None
         shared_path = None
+        self.pod_gc = pod_gc
         if shared_directory:
             if shared_directory is True:
                 self.volumes = ['/mnt/shared']
@@ -126,6 +127,10 @@ def spec(self):
             'tolerations': [V1Toleration(key='cloudharness/temporary-job', operator='Equal', value='true').to_dict()],
             'serviceAccountName': SERVICE_ACCOUNT,
             'imagePullSecrets': [{'name': config.CloudharnessConfig.get_registry_secret()}],
+            'podGC': self.pod_gc or {
+                'strategy': 'OnWorkflowSuccess',
+                'deleteDelayDuration': self.ttl_strategy['secondsAfterSuccess'] if self.ttl_strategy else "600s",
+            },
             'volumes': [{
                 # mount allvalues so we can use the cloudharness Python library
                 'name': 'cloudharness-allvalues',
@@ -168,7 +173,7 @@ def add_on_exit_notify_handler(self, spec):
         payload = self.on_exit_notify['payload']
         exit_task = CustomTask(
             name="exit-handler",
-            image_name='workflows-notify-queue',
+            image_name=self.on_exit_notify.get('image', 'workflows-notify-queue'),
             workflow_result='{{workflow.status}}',
             queue_name=queue,
             payload=payload
@@ -206,7 +211,7 @@ def submit(self):
         log.debug("Submitting workflow\n" + yaml.dump(op))
 
         # TODO use rest api for that? Include this into cloudharness.workflows?
-        self.persisted = argo.submit_workflow(op)
+        self.persisted = argo_service.submit_workflow(op)
         return self.persisted
 
     def is_running(self):
@@ -216,7 +221,7 @@ def is_running(self):
         return False
 
     def refresh(self):
-        self.persisted = argo.get_workflow(self.persisted.name)
+        self.persisted = argo_service.get_workflow(self.persisted.name)
 
     def is_error(self):
         if self.persisted:
@@ -313,7 +318,7 @@ def execute(self, timeout=None):
         while not self.persisted.is_finished():
             time.sleep(POLLING_WAIT_SECONDS)
             log.debug(f"Polling argo workflow {self.persisted.name}")
-            self.persisted = argo.get_workflow(self.persisted.name)
+            self.persisted = argo_service.get_workflow(self.persisted.name)
             log.debug(f"Polling succeeded for \
                {self.persisted.name}. Current phase: {self.persisted.status}")
             if timeout and time.time() - start_time > timeout:

libraries/cloudharness-common/cloudharness/workflows/tasks.py

@@ -1,25 +1,29 @@
-from . import argo
-
+import argo.workflows
+from cloudharness.utils import dict_merge
+from . import argo_service
+from argo.workflows.client import V1alpha1Template
 from cloudharness.utils.env import get_cloudharness_variables, get_image_full_tag
 from .utils import WORKFLOW_NAME_VARIABLE_NAME, PodExecutionContext, affinity_spec, is_accounts_present, volume_mount_template, volume_requires_affinity
 
 SERVICE_ACCOUNT = 'argo-workflows'
 
 
-class Task(argo.ArgoObject):
+class Task(argo_service.ArgoObject):
     """
     Abstract interface for a task.
     """
 
-    def __init__(self, name, resources={}, volume_mounts=[], **env_args):
+    def __init__(self, name, resources=None, volume_mounts=None, retry_limit=10, template_overrides: V1alpha1Template = None, **env_args):
         self.name = name.replace(' ', '-').lower()
-        self.resources = resources
+        self.resources = resources or {}
         self.__envs = get_cloudharness_variables()
-        self.volume_mounts = volume_mounts
+        self.volume_mounts = volume_mounts or []
         self.external_volumes = [
             v.split(':')[0] for v in self.volume_mounts if volume_requires_affinity(v)]
         for k in env_args:
             self.__envs[k] = str(env_args[k])
+        self.retry_limit = retry_limit
+        self.template_overrides = template_overrides.to_dict() if template_overrides else {}
 
     def metadata_spec(self):
         return {
@@ -85,14 +89,14 @@ def volumes_mounts_spec(self):
 
 class ContainerizedTask(Task):
 
-    def __init__(self, name, resources={}, image_pull_policy='IfNotPresent', command=None, **env_args):
-        super().__init__(name, resources, **env_args)
+    def __init__(self, name, resources={}, image_pull_policy='IfNotPresent', command=None, retry_limit=10, template_overrides: V1alpha1Template = None, **env_args):
+        super().__init__(name, resources, retry_limit=retry_limit, template_overrides=template_overrides, **env_args)
         self.image_pull_policy = image_pull_policy
         self.command = command
 
     def spec(self):
 
-        spec = {
+        spec = dict_merge({
             'container': {
                 'image': self.image_name,
                 'env': self.envs,
@@ -104,9 +108,12 @@ def spec(self):
             'metadata': self.metadata_spec(),
             'name': self.name,
             'outputs': {},
-            'affinity': self.affinity_spec()
+            'affinity': self.affinity_spec(),
+            'retryStrategy': {
+                'limit': self.retry_limit
+            }
 
-        }
+        }, self.template_overrides, merge_none=False)
         if self.command is not None:
             spec['container']['command'] = self.command
         return spec
@@ -123,7 +130,7 @@ def __init__(self, name, source, **kwargs):
 
     def spec(self):
 
-        return {
+        return dict_merge({
             'name': self.name,
             'affinity': self.affinity_spec(),
             'metadata': self.metadata_spec(),
@@ -135,18 +142,18 @@ def spec(self):
                     'volumeMounts': self.volumes_mounts_spec(),
                     'command': [self.command]
             }
-        }
+        }, self.template_overrides, merge_none=False)
 
     @property
     def command(self):
         raise NotImplemented
 
 
 class PythonTask(InlinedTask):
-    def __init__(self, name, func):
+    def __init__(self, name, func, **kwargs):
         import inspect
         super().__init__(name, (inspect.getsource(
-            func) + f"\n{func.__name__}()").strip())
+            func) + f"\n{func.__name__}()").strip(), **kwargs)
 
     @property
     def image_name(self):

libraries/cloudharness-common/tests/test_applications.py

@@ -1,8 +1,7 @@
-from cloudharness.utils.config import CloudharnessConfig, ConfigObject
 from cloudharness.applications import ApplicationConfiguration, get_configuration
-from .test_env import set_default_environment
-
-set_default_environment()
+from cloudharness.utils.config import CloudharnessConfig, ConfigObject
+from .test_env import set_test_environment
+set_test_environment()
 
 
 conf_1 = {

libraries/cloudharness-common/tests/test_env.py

@@ -1,9 +1,7 @@
-from cloudharness.utils.config import CloudharnessConfig as conf
 from cloudharness.utils.env import set_default_environment
-import pytest
+from cloudharness.utils.config import CloudharnessConfig as conf
 import os
 import yaml
-
 HERE = os.path.dirname(os.path.realpath(__file__))
 os.environ["CH_VALUES_PATH"] = os.path.join(HERE, "values.yaml")