From b4252724e5e5dc0088828778d36ae767f3608f23 Mon Sep 17 00:00:00 2001
From: Maximilian Friedersdorff <max@friedersdorff.com>
Date: Thu, 24 Nov 2016 13:19:48 +0000
Subject: [PATCH 1/7] Estimate the walltime usage over the last 30 days.

Log at debug level if usage is over 20 seconds.
---
 lib/galaxy/jobs/handler.py | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py
index b408dfd3d3b..5f3677e495e 100644
--- a/lib/galaxy/jobs/handler.py
+++ b/lib/galaxy/jobs/handler.py
@@ -6,6 +6,7 @@ import os
 import time
 import logging
 import threading
+import datetime
 from Queue import Queue, Empty
 
 from sqlalchemy.sql.expression import and_, or_, select, func, true, null
@@ -375,6 +376,33 @@ class JobHandlerQueue( object ):
         # job is ready to run, check limits
         # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable
         state = self.__check_destination_jobs( job, job_wrapper )
+
+        # Check 
+        jobs_to_check = self.sa_session.query( model.Job ).filter(
+            model.Job.user_id == job.user.id,
+            model.Job.update_time >=
+                datetime.datetime.now() - datetime.timedelta(30),
+            model.Job.state == 'ok'
+        ).all()
+        time_spent = datetime.timedelta(0)
+        for job in jobs_to_check:
+            # History is job.state_history
+            started = None 
+            finished = None
+            for history in sorted(
+                    job.state_history,
+                    key=lambda history: history.update_time):
+                if history.state == "running":
+                    started = history.create_time
+                elif history.state == "ok":
+                    finished = history.create_time
+
+            time_spent += finished - started
+
+        if time_spent > datetime.timedelta(seconds=20):
+            log.debug("Used too much cpu time")
+                    
+
         if state == JOB_READY:
             state = self.__check_user_jobs( job, job_wrapper )
         if state == JOB_READY and self.app.config.enable_quotas:

From 2408502c2d316869df0f84b65e50db3114f9529a Mon Sep 17 00:00:00 2001
From: Maximilian Friedersdorff <max@friedersdorff.com>
Date: Thu, 24 Nov 2016 16:59:54 +0000
Subject: [PATCH 2/7] Add total_walltime limit type

Add to sample config, config reader and unittest
---
 config/job_conf.xml.sample_advanced      |  7 +++++++
 lib/galaxy/jobs/__init__.py              | 16 ++++++++++++++++
 test/unit/jobs/test_job_configuration.py |  3 +++
 3 files changed, 26 insertions(+)

diff --git a/config/job_conf.xml.sample_advanced b/config/job_conf.xml.sample_advanced
index 4192008e45e..ca8c22aae33 100644
--- a/config/job_conf.xml.sample_advanced
+++ b/config/job_conf.xml.sample_advanced
@@ -762,6 +762,13 @@
                 will be terminated by Galaxy.
          -->
         <limit type="walltime">24:00:00</limit>
+	<!-- total_walltime:
+		Total walltime that jobs may not exceed during during a
+		set period.  If total walltime of finished jobs exceeds
+		this value, any new jobs are paused.  window is a number
+		in days, representing the period.
+         -->
+        <limit type="total_walltime" window="30">24:00:00</limit>
         <!-- output_size:
                 Size that any defined tool output can grow to before the job
                 will be terminated. This does not include temporary files
diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 4aa18d54c11..dfc302a7499 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -265,12 +265,14 @@ class JobConfiguration( object ):
         types = dict(registered_user_concurrent_jobs=int,
                      anonymous_user_concurrent_jobs=int,
                      walltime=str,
+                     total_walltime=str,
                      output_size=util.size_to_bytes)
 
         self.limits = Bunch(registered_user_concurrent_jobs=None,
                             anonymous_user_concurrent_jobs=None,
                             walltime=None,
                             walltime_delta=None,
+                            total_walltime={},
                             output_size=None,
                             destination_user_concurrent_jobs={},
                             destination_total_concurrent_jobs={})
@@ -287,6 +289,13 @@ class JobConfiguration( object ):
                         self.limits.destination_total_concurrent_jobs[id] = int(limit.text)
                     else:
                         self.limits.destination_user_concurrent_jobs[id] = int(limit.text)
+                elif type == 'total_walltime':
+                    self.limits.total_walltime["window"] = (
+                        int(limit.get('window')) or 30
+                    )
+                    self.limits.total_walltime["raw"] = (
+                        types.get(type, str)(limit.text)
+                    )
                 elif limit.text:
                     self.limits.__dict__[type] = types.get(type, str)(limit.text)
 
@@ -294,6 +303,13 @@ class JobConfiguration( object ):
             h, m, s = [ int( v ) for v in self.limits.walltime.split( ':' ) ]
             self.limits.walltime_delta = datetime.timedelta( 0, s, 0, 0, m, h )
 
+        if "raw" in self.limits.total_walltime:
+            h, m, s = [ int( v ) for v in
+                        self.limits.total_walltime["raw"].split( ':' ) ]
+            self.limits.total_walltime["delta"] = datetime.timedelta(
+                0, s, 0, 0, m, h
+            )
+
         log.debug('Done loading job configuration')
 
     def __parse_job_conf_legacy(self):
diff --git a/test/unit/jobs/test_job_configuration.py b/test/unit/jobs/test_job_configuration.py
index 2a178a07bcf..9045f768a35 100644
--- a/test/unit/jobs/test_job_configuration.py
+++ b/test/unit/jobs/test_job_configuration.py
@@ -100,6 +100,7 @@ class JobConfXmlParserTestCase( unittest.TestCase ):
         assert limits.anonymous_user_concurrent_jobs is None
         assert limits.walltime is None
         assert limits.walltime_delta is None
+        assert limits.total_walltime == {}
         assert limits.output_size is None
         assert limits.destination_user_concurrent_jobs == {}
         assert limits.destination_total_concurrent_jobs == {}
@@ -113,6 +114,8 @@ class JobConfXmlParserTestCase( unittest.TestCase ):
         assert limits.destination_user_concurrent_jobs[ "mycluster" ] == 2
         assert limits.destination_user_concurrent_jobs[ "longjobs" ] == 1
         assert limits.walltime_delta == datetime.timedelta( 0, 0, 0, 0, 0, 24 )
+        assert limits.total_walltime["delta"] == datetime.timedelta( 0, 0, 0, 0, 0, 24)
+        assert limits.total_walltime["window"] == 30
 
     def test_env_parsing( self ):
         self.__with_advanced_config()

From 8e4072179789517427727339e3e9a2e911fae6a6 Mon Sep 17 00:00:00 2001
From: Maximilian Friedersdorff <max@friedersdorff.com>
Date: Thu, 24 Nov 2016 17:01:09 +0000
Subject: [PATCH 3/7] Use conf for walltime limit and window.  Pause job if
 necessary

---
 lib/galaxy/jobs/handler.py | 66 +++++++++++++++++++++-----------------
 1 file changed, 37 insertions(+), 29 deletions(-)

diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py
index 5f3677e495e..f0206fd2d83 100644
--- a/lib/galaxy/jobs/handler.py
+++ b/lib/galaxy/jobs/handler.py
@@ -19,7 +19,7 @@ from galaxy.jobs.mapper import JobNotReadyException
 log = logging.getLogger( __name__ )
 
 # States for running a job. These are NOT the same as data states
-JOB_WAIT, JOB_ERROR, JOB_INPUT_ERROR, JOB_INPUT_DELETED, JOB_READY, JOB_DELETED, JOB_ADMIN_DELETED, JOB_USER_OVER_QUOTA = 'wait', 'error', 'input_error', 'input_deleted', 'ready', 'deleted', 'admin_deleted', 'user_over_quota'
+JOB_WAIT, JOB_ERROR, JOB_INPUT_ERROR, JOB_INPUT_DELETED, JOB_READY, JOB_DELETED, JOB_ADMIN_DELETED, JOB_USER_OVER_QUOTA, JOB_USER_OVER_TOTAL_WALLTIME = 'wait', 'error', 'input_error', 'input_deleted', 'ready', 'deleted', 'admin_deleted', 'user_over_quota', 'user_over_total_walltime'
 DEFAULT_JOB_PUT_FAILURE_MESSAGE = 'Unable to run job due to a misconfiguration of the Galaxy job running system.  Please contact a site administrator.'
 
 
@@ -285,8 +285,13 @@ class JobHandlerQueue( object ):
                     log.info( "(%d) Job deleted by user while still queued" % job.id )
                 elif job_state == JOB_ADMIN_DELETED:
                     log.info( "(%d) Job deleted by admin while still queued" % job.id )
-                elif job_state == JOB_USER_OVER_QUOTA:
-                    log.info( "(%d) User (%s) is over quota: job paused" % ( job.id, job.user_id ) )
+                elif job_state in (JOB_USER_OVER_QUOTA,
+                                   JOB_USER_OVER_TOTAL_WALLTIME):
+                    if job_state == JOB_USER_OVER_QUOTA:
+                        log.info( "(%d) User (%s) is over quota: job paused" % ( job.id, job.user_id ) )
+                    else:
+                        log.info(  "(%d) User (%s) is over total walltime limit: job paused" % ( job.id, job.user_id ) )
+
                     job.set_state( model.Job.states.PAUSED )
                     for dataset_assoc in job.output_datasets + job.output_library_datasets:
                         dataset_assoc.dataset.dataset.state = model.Dataset.states.PAUSED
@@ -377,32 +382,6 @@ class JobHandlerQueue( object ):
         # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable
         state = self.__check_destination_jobs( job, job_wrapper )
 
-        # Check 
-        jobs_to_check = self.sa_session.query( model.Job ).filter(
-            model.Job.user_id == job.user.id,
-            model.Job.update_time >=
-                datetime.datetime.now() - datetime.timedelta(30),
-            model.Job.state == 'ok'
-        ).all()
-        time_spent = datetime.timedelta(0)
-        for job in jobs_to_check:
-            # History is job.state_history
-            started = None 
-            finished = None
-            for history in sorted(
-                    job.state_history,
-                    key=lambda history: history.update_time):
-                if history.state == "running":
-                    started = history.create_time
-                elif history.state == "ok":
-                    finished = history.create_time
-
-            time_spent += finished - started
-
-        if time_spent > datetime.timedelta(seconds=20):
-            log.debug("Used too much cpu time")
-                    
-
         if state == JOB_READY:
             state = self.__check_user_jobs( job, job_wrapper )
         if state == JOB_READY and self.app.config.enable_quotas:
@@ -414,6 +393,35 @@ class JobHandlerQueue( object ):
                         return JOB_USER_OVER_QUOTA, job_destination
                 except AssertionError as e:
                     pass  # No history, should not happen with an anon user
+        # Check total walltime limits
+        if (state == JOB_READY and
+            "delta" in self.app.job_config.limits.total_walltime):
+            jobs_to_check = self.sa_session.query( model.Job ).filter(
+                model.Job.user_id == job.user.id,
+                model.Job.update_time >= datetime.datetime.now() -
+                datetime.timedelta(
+                    self.app.job_config.limits.total_walltime["window"]
+                ),
+                model.Job.state == 'ok'
+            ).all()
+            time_spent = datetime.timedelta(0)
+            for job in jobs_to_check:
+                # History is job.state_history
+                started = None
+                finished = None
+                for history in sorted(
+                        job.state_history,
+                        key=lambda history: history.update_time):
+                    if history.state == "running":
+                        started = history.create_time
+                    elif history.state == "ok":
+                        finished = history.create_time
+
+                time_spent += finished - started
+
+            if time_spent > self.app.job_config.limits.total_walltime["delta"]:
+                return JOB_USER_OVER_TOTAL_WALLTIME, job_destination
+
         return state, job_destination
 
     def __verify_in_memory_job_inputs( self, job ):

From 3d4de14840847636ad81c36bd64449fb959220b9 Mon Sep 17 00:00:00 2001
From: Maximilian Friedersdorff <max@friedersdorff.com>
Date: Thu, 24 Nov 2016 17:06:08 +0000
Subject: [PATCH 4/7] Comply with galaxy style guide regarding whitespace.

---
 lib/galaxy/jobs/__init__.py |  2 +-
 lib/galaxy/jobs/handler.py  | 12 ++++++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index dfc302a7499..9c88f296480 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -291,7 +291,7 @@ class JobConfiguration( object ):
                         self.limits.destination_user_concurrent_jobs[id] = int(limit.text)
                 elif type == 'total_walltime':
                     self.limits.total_walltime["window"] = (
-                        int(limit.get('window')) or 30
+                        int( limit.get('window') ) or 30
                     )
                     self.limits.total_walltime["raw"] = (
                         types.get(type, str)(limit.text)
diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py
index f0206fd2d83..60fbb636b62 100644
--- a/lib/galaxy/jobs/handler.py
+++ b/lib/galaxy/jobs/handler.py
@@ -285,12 +285,12 @@ class JobHandlerQueue( object ):
                     log.info( "(%d) Job deleted by user while still queued" % job.id )
                 elif job_state == JOB_ADMIN_DELETED:
                     log.info( "(%d) Job deleted by admin while still queued" % job.id )
-                elif job_state in (JOB_USER_OVER_QUOTA,
-                                   JOB_USER_OVER_TOTAL_WALLTIME):
+                elif job_state in ( JOB_USER_OVER_QUOTA,
+                                    JOB_USER_OVER_TOTAL_WALLTIME ):
                     if job_state == JOB_USER_OVER_QUOTA:
                         log.info( "(%d) User (%s) is over quota: job paused" % ( job.id, job.user_id ) )
                     else:
-                        log.info(  "(%d) User (%s) is over total walltime limit: job paused" % ( job.id, job.user_id ) )
+                        log.info( "(%d) User (%s) is over total walltime limit: job paused" % ( job.id, job.user_id ) )
 
                     job.set_state( model.Job.states.PAUSED )
                     for dataset_assoc in job.output_datasets + job.output_library_datasets:
@@ -394,8 +394,8 @@ class JobHandlerQueue( object ):
                 except AssertionError as e:
                     pass  # No history, should not happen with an anon user
         # Check total walltime limits
-        if (state == JOB_READY and
-            "delta" in self.app.job_config.limits.total_walltime):
+        if ( state == JOB_READY and
+             "delta" in self.app.job_config.limits.total_walltime ):
             jobs_to_check = self.sa_session.query( model.Job ).filter(
                 model.Job.user_id == job.user.id,
                 model.Job.update_time >= datetime.datetime.now() -
@@ -411,7 +411,7 @@ class JobHandlerQueue( object ):
                 finished = None
                 for history in sorted(
                         job.state_history,
-                        key=lambda history: history.update_time):
+                        key=lambda history: history.update_time ):
                     if history.state == "running":
                         started = history.create_time
                     elif history.state == "ok":

From db1b51a0802e392787c3a71e7a8bd202b568bf3a Mon Sep 17 00:00:00 2001
From: Maximilian Friedersdorff <max@friedersdorff.com>
Date: Fri, 25 Nov 2016 13:35:34 +0000
Subject: [PATCH 5/7] Fix errors in documentation for total_walltime limit

---
 config/job_conf.xml.sample_advanced | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/config/job_conf.xml.sample_advanced b/config/job_conf.xml.sample_advanced
index ca8c22aae33..fedd7deaec9 100644
--- a/config/job_conf.xml.sample_advanced
+++ b/config/job_conf.xml.sample_advanced
@@ -763,10 +763,10 @@
          -->
         <limit type="walltime">24:00:00</limit>
 	<!-- total_walltime:
-		Total walltime that jobs may not exceed during during a
-		set period.  If total walltime of finished jobs exceeds
-		this value, any new jobs are paused.  window is a number
-		in days, representing the period.
+		Total walltime that jobs may not exceed during a set period.
+		If total walltime of finished jobs exceeds this value, any
+		new jobs are paused.  `window` is a number in days,
+		representing the period.
          -->
         <limit type="total_walltime" window="30">24:00:00</limit>
         <!-- output_size:

From 91e65aa42c8e2439883da1d690e770a5a9df1a79 Mon Sep 17 00:00:00 2001
From: Maximilian Friedersdorff <max@friedersdorff.com>
Date: Fri, 25 Nov 2016 13:36:03 +0000
Subject: [PATCH 6/7] Move datetime import to the top of the file.

---
 lib/galaxy/jobs/handler.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py
index 60fbb636b62..4a4de2c1b52 100644
--- a/lib/galaxy/jobs/handler.py
+++ b/lib/galaxy/jobs/handler.py
@@ -2,11 +2,11 @@
 Galaxy job handler, prepares, runs, tracks, and finishes Galaxy jobs
 """
 
+import datetime
 import os
 import time
 import logging
 import threading
-import datetime
 from Queue import Queue, Empty
 
 from sqlalchemy.sql.expression import and_, or_, select, func, true, null

From f467bcb70697bdaac0f94f187970b98f9b58bfe9 Mon Sep 17 00:00:00 2001
From: Maximilian Friedersdorff <max@friedersdorff.com>
Date: Mon, 28 Nov 2016 11:11:56 +0000
Subject: [PATCH 7/7] Add total_walltime to legacy job conf

---
 lib/galaxy/jobs/__init__.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 9c88f296480..3fa172a1b48 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -372,6 +372,7 @@ class JobConfiguration( object ):
                             anonymous_user_concurrent_jobs=self.app.config.anonymous_user_job_limit,
                             walltime=self.app.config.job_walltime,
                             walltime_delta=self.app.config.job_walltime_delta,
+                            total_walltime={},
                             output_size=self.app.config.output_size_limit,
                             destination_user_concurrent_jobs={},
                             destination_total_concurrent_jobs={})