From 50257aac80c7aedf41ad436d475640d586ca732c Mon Sep 17 00:00:00 2001
From: John Chilton <jmchilton@gmail.com>
Date: Thu, 15 Dec 2016 14:18:21 -0500
Subject: [PATCH 1/2] Add re-submission info to job_conf.xml.sample_advanced.

---
 config/job_conf.xml.sample_advanced | 43 +++++++++++++++++++++--------
 1 file changed, 31 insertions(+), 12 deletions(-)

diff --git a/config/job_conf.xml.sample_advanced b/config/job_conf.xml.sample_advanced
index fedd7deaec9..fe09304cf3d 100644
--- a/config/job_conf.xml.sample_advanced
+++ b/config/job_conf.xml.sample_advanced
@@ -574,22 +574,36 @@
             <!-- <param id="docker_default_container_id">busybox:ubuntu-14.04</param> -->
         </destination>
 
-        <!-- Jobs that hit the walltime on one destination can be automatically
-             resubmitted to another destination. Walltime detection is
-             currently only implemented in the slurm runner.
+        <!-- Jobs can be re-submitted for various reasons (to the same destination or others,
+             with or without a short delay). For instance, jobs that hit the walltime on one
+             destination can be automatically resubmitted to another destination. Re-submission
+             is defined on a per-destination basis using ``resubmit`` tags. Re-submission only
+             happens currently in response to problems in the job runner - so for instance if a
+             job fails to allocate memory but the job runner doesn't detect this and completes
+             the job normally but the exit code indicates the error - the job failure
+             re-submission won't run yet (this will be added in the future).
 
-             Multiple resubmit tags can be defined, the first resubmit matching
-             the terminal condition of a job will be used.
+             Multiple `resubmit` tags can be defined, the first resubmit condition that is true
+             (i.e. evaluates to a Python truthy value) will be used for a particular job failure.
 
-             The 'condition' attribute is optional, if not present, the
-             resubmit destination will be used for all conditions. The
-             conditions currently implemented are:
+             The ``condition`` attribute is optional, if not present, the
+             resubmit destination will be used for all relevant failure types.
+             Conditions are expressed as Python-like expressions (a fairly safe subset of Python
+             is available). These expressions include math and logical operators, numbers,
+             strings, etc.... The following variables are available in these expressions:
 
-               - "walltime_reached"
-               - "memory_limit_reached"
+               - "walltime_reached" (True if and only if the job runner indicates a walltime maximum was reached)
+               - "memory_limit_reached" (True if and only if the job runner indicates a memory limit was hit)
+               - "unknown_error" (True for job or job runner problems that aren't otherwise classified)
+               - "attempt" (the re-submission attempt number this is)
+               - "seconds_since_queued" (the number of seconds since the last time the job was in a queued state within Galaxy)
+               - "seconds_running" (the number of seconds the job was in a running state within Galaxy)
 
-             The 'handler' tag is optional, if not present, the job's original
-             handler will be reused for the resubmitted job.
+             The ``handler`` attribute is optional, if not present, the job's original
+             handler will be reused for the resubmitted job. The ``destination`` attriubte
+             is optional, if not present the job's original destination will be reused for the
+             re-submission. The ``delay`` attribute is optional, if present it will cause the job to
+             delay for that number of seconds before being re-submitted. 
         -->
         <destination id="short_fast" runner="slurm">
             <param id="nativeSpecification">--time=00:05:00 --nodes=1</param>
@@ -603,6 +617,11 @@
             <param id="nativeSpecification">--mem-per-cpu=512</param>
             <resubmit condition="memory_limit_reached" destination="bigmem" />
         </destination>
+        <destination id="retry_on_unknown_problems" runner="slurm">
+            <!-- Just retry the job 5 times if un-categories errors occur backing
+                 off by 30 more seconds between attempts. -->
+            <resubmit condition="unknown_error and attempt &lt;= 5" delay="attempt * 30" />
+        </destination>
         <!-- Any tag param in this file can be set using an environment variable or using
              values from galaxy.ini using the from_environ and from_config attributes
              repectively. The text of the param will still be used if that environment variable

From 34f0b0fe8911c01562c7dfc0e83e0d56df50f66c Mon Sep 17 00:00:00 2001
From: John Chilton <jmchilton@gmail.com>
Date: Thu, 15 Dec 2016 16:11:55 -0500
Subject: [PATCH 2/2] Implement a Galaxy config option for default job
 re-submission conditions.

---
 config/galaxy.ini.sample                      | 14 ++++++
 lib/galaxy/config.py                          |  5 +++
 lib/galaxy/jobs/__init__.py                   | 19 +++++++-
 .../resubmission_default_job_conf.xml         | 43 +++++++++++++++++++
 test/integration/resubmission_runners.py      | 20 +++++++++
 test/integration/test_job_resubmission.py     | 39 ++++++++++++-----
 test/unit/jobs/test_job_configuration.py      |  1 +
 7 files changed, 130 insertions(+), 11 deletions(-)
 create mode 100644 test/integration/resubmission_default_job_conf.xml

diff --git a/config/galaxy.ini.sample b/config/galaxy.ini.sample
index 8028f9917c3..d92b517788d 100644
--- a/config/galaxy.ini.sample
+++ b/config/galaxy.ini.sample
@@ -1148,6 +1148,20 @@ use_interactive = True
 # be configured through the job configuration file.
 #job_config_file = config/job_conf.xml
 
+# When jobs fail due to job runner problems, Galaxy can be configured to retry
+# these or reroute the jobs to new destinations. Very fine control of this is
+# available with resubmit declarations in job_conf.xml. For simple deployments
+# of Galaxy though, the following attribute can define resubmission conditions
+# for all job destinations. If any job destination defines even one
+# resubmission condition explicitly in job_conf.xml - the condition described
+# by this option will not apply to that destination. For instance, the condition:
+# 'attempt < 3 and unknown_error and (time_running < 300 or time_since_queued < 300)'
+# would retry up to two times jobs that didn't fail due to detected memory or
+# walltime limits but did fail quickly (either while queueing or running). The
+# commented out default below results in no default job resubmission condition,
+# failing jobs are just failed outright.
+#default_job_resubmission_condition = 
+
 # In multiprocess configurations, notification between processes about new jobs
 # must be done via the database.  In single process configurations, this can be
 # done in memory, which is a bit quicker.
diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py
index 7d6a754e772..b7c30417c6c 100644
--- a/lib/galaxy/config.py
+++ b/lib/galaxy/config.py
@@ -356,6 +356,11 @@ class Configuration( object ):
         self.involucro_path = resolve_path(involucro_path, self.root)
         self.involucro_auto_init = string_as_bool(kwargs.get( 'involucro_auto_init', True))
 
+        default_job_resubmission_condition = kwargs.get( 'default_job_resubmission_condition', '')
+        if not default_job_resubmission_condition.strip():
+            default_job_resubmission_condition = None
+        self.default_job_resubmission_condition = default_job_resubmission_condition
+
         # Configuration options for taking advantage of nginx features
         self.upstream_gzip = string_as_bool( kwargs.get( 'upstream_gzip', False ) )
         self.apache_xsendfile = string_as_bool( kwargs.get( 'apache_xsendfile', False ) )
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 5a89e46afa2..342d4981c3c 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -138,6 +138,17 @@ class JobConfiguration( object ):
         self.resource_parameters = {}
         self.limits = Bunch()
 
+        default_resubmits = []
+        default_resubmit_condition = self.app.config.default_job_resubmission_condition
+        if default_resubmit_condition:
+            default_resubmits.append(dict(
+                destination=None,
+                condition=default_resubmit_condition,
+                handler=None,
+                delay=None,
+            ))
+        self.default_resubmits = default_resubmits
+
         self.__parse_resource_parameters()
         # Initialize the config
         job_config_file = self.app.config.job_config_file
@@ -230,7 +241,13 @@ class JobConfiguration( object ):
             job_destination = JobDestination(**dict(destination.items()))
             job_destination['params'] = self.__get_params(destination)
             job_destination['env'] = self.__get_envs(destination)
-            job_destination['resubmit'] = self.__get_resubmits(destination)
+            destination_resubmits = self.__get_resubmits(destination)
+            if destination_resubmits:
+                resubmits = self.default_resubmits
+            else:
+                resubmits = self.default_resubmits
+            job_destination["resubmit"] = resubmits
+
             self.destinations[id] = (job_destination,)
             if job_destination.tags is not None:
                 for tag in job_destination.tags:
diff --git a/test/integration/resubmission_default_job_conf.xml b/test/integration/resubmission_default_job_conf.xml
new file mode 100644
index 00000000000..a09e58e11af
--- /dev/null
+++ b/test/integration/resubmission_default_job_conf.xml
@@ -0,0 +1,43 @@
+<?xml version="1.0"?>
+<!-- 
+    Slimmed down resubmission_job_conf.xml for testing default resubmission rules.
+-->
+<job_conf>
+    <plugins>
+        <plugin id="local" type="runner" load="galaxy.jobs.runners.local:LocalJobRunner" workers="2"/>
+        <plugin id="first_failure_runner" type="runner" load="integration.resubmission_runners:FailOnlyFirstJobRunner" workers="2">
+        </plugin>
+        <plugin id="dynamic" type="runner">
+            <param id="rules_module">integration.resubmission_rules</param>
+        </plugin>
+    </plugins>
+
+    <handlers>
+        <handler id="main"/>
+    </handlers>
+
+    <destinations default="initial_destination">
+        <destination id="initial_destination" runner="dynamic">
+            <param id="type">python</param>
+            <param id="function">initial_destination</param>
+        </destination>
+
+        <destination id="fail_first_try" runner="first_failure_runner">
+        </destination>
+
+        <!-- Upload destination. -->
+        <destination id="local" runner="local">
+        </destination>
+
+    </destinations>
+
+    <resources default="test">
+      <group id="upload"></group>
+      <group id="test">test_name,failure_state,initial_destination,run_for</group>
+    </resources>
+
+    <tools>
+        <tool id="upload1" destination="local" resources="upload" />
+    </tools>
+
+</job_conf>
diff --git a/test/integration/resubmission_runners.py b/test/integration/resubmission_runners.py
index 39cbb7c01f8..572fece61fe 100644
--- a/test/integration/resubmission_runners.py
+++ b/test/integration/resubmission_runners.py
@@ -63,4 +63,24 @@ class AssertionJobRunner(LocalJobRunner):
         super(AssertionJobRunner, self).queue_job(job_wrapper)
 
 
+class FailOnlyFirstJobRunner(LocalJobRunner):
+    """Job runner that knows about test cases and checks final state assumptions."""
+
+    tests_seen = []
+
+    def queue_job(self, job_wrapper):
+        resource_parameters = job_wrapper.get_resource_parameters()
+        try:
+            test_name = resource_parameters["test_name"]
+        except KeyError:
+            job_wrapper.fail("Job resource parameter test_name not set as required for this job runner.")
+            return
+
+        if test_name in self.tests_seen:
+            super(FailOnlyFirstJobRunner, self).queue_job(job_wrapper)
+        else:
+            self.tests_seen.append(test_name)
+            self._fail_job_local(job_wrapper, "Failing first attempt")
+
+
 __all__ = ('FailsJobRunner', 'AssertionJobRunner')
diff --git a/test/integration/test_job_resubmission.py b/test/integration/test_job_resubmission.py
index cefe368d36e..cffbdb5e5e8 100644
--- a/test/integration/test_job_resubmission.py
+++ b/test/integration/test_job_resubmission.py
@@ -6,10 +6,27 @@ from base import integration_util
 
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
 JOB_RESUBMISSION_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "resubmission_job_conf.xml")
+JOB_RESUBMISSION_DEFAULT_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "resubmission_default_job_conf.xml")
 JOB_RESUBMISSION_JOB_RESOURCES_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "resubmission_job_resource_parameters_conf.xml")
 
 
-class JobResubmissionIntegrationTestCase(integration_util.IntegrationTestCase):
+class _BaseResubmissionIntegerationTestCase(integration_util.IntegrationTestCase):
+    framework_tool_and_types = True
+
+    def _assert_job_passes(self, resource_parameters):
+        self._run_tool_test("simple_constructs", resource_parameters=resource_parameters)
+
+    def _assert_job_fails(self, resource_parameters):
+        exception_thrown = False
+        try:
+            self._run_tool_test("simple_constructs", resource_parameters=resource_parameters)
+        except Exception:
+            exception_thrown = True
+
+        assert exception_thrown
+
+
+class JobResubmissionIntegrationTestCase(_BaseResubmissionIntegerationTestCase):
 
     framework_tool_and_types = True
 
@@ -85,14 +102,16 @@ class JobResubmissionIntegrationTestCase(integration_util.IntegrationTestCase):
                                                      "initial_destination": "resubmit_after_two_delays",
                                                      "failure_state": "unknown_error"})
 
-    def _assert_job_passes(self, resource_parameters):
-        self._run_tool_test("simple_constructs", resource_parameters=resource_parameters)
 
-    def _assert_job_fails(self, resource_parameters):
-        exception_thrown = False
-        try:
-            self._run_tool_test("simple_constructs", resource_parameters=resource_parameters)
-        except Exception:
-            exception_thrown = True
+class JobResubmissionDefaultIntegrationTestCase(_BaseResubmissionIntegerationTestCase):
 
-        assert exception_thrown
+    framework_tool_and_types = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["default_job_resubmission_condition"] = "attempt < 2"
+        config["job_config_file"] = JOB_RESUBMISSION_DEFAULT_JOB_CONFIG_FILE
+        config["job_resource_params_file"] = JOB_RESUBMISSION_JOB_RESOURCES_CONFIG_FILE
+
+    def test_default_resubmission(self):
+        self._assert_job_passes(resource_parameters={"test_name": "test_default_resubmission"})
diff --git a/test/unit/jobs/test_job_configuration.py b/test/unit/jobs/test_job_configuration.py
index 9045f768a35..ec19701b158 100644
--- a/test/unit/jobs/test_job_configuration.py
+++ b/test/unit/jobs/test_job_configuration.py
@@ -22,6 +22,7 @@ class JobConfXmlParserTestCase( unittest.TestCase ):
             use_tasked_jobs=False,
             job_resource_params_file="/tmp/fake_absent_path",
             config_dict={},
+            default_job_resubmission_condition="",
         )
         self.__write_config_from( SIMPLE_JOB_CONF )
         self.app = bunch.Bunch( config=self.config, job_metrics=MockJobMetrics() )