Add helper for removing old jobs

1 files changed, 153 insertions, 0 deletions

diff --git a/lorry-controller-remove-old-jobs b/lorry-controller-remove-old-jobs
new file mode 100755
index 0000000..1448649
--- /dev/null
+++ b/lorry-controller-remove-old-jobs
@@ -0,0 +1,153 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2014  Codethink Limited
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+
+import json
+import logging
+import time
+import urllib2
+import urlparse
+import contextlib
+
+import cliapp
+
+
+class JobInfo(object):
+
+    def __init__(self, job_id, exit_code, exit_timestamp):
+        self.job_id = job_id
+        self.exit_code = exit_code
+        self.exit_timestamp = exit_timestamp
+
+    def __repr__(self):
+        return 'JobInfo(%s,%s,%s)' % (
+            self.job_id, self.exit_code, self.exit_timestamp)
+
+
+class OldJobRemover(cliapp.Application):
+
+    def add_settings(self):
+        self.settings.string(
+            ['webapp-host'],
+            'address of WEBAPP',
+            default='localhost')
+
+        self.settings.integer(
+            ['webapp-port'],
+            'port of WEBAPP',
+            default=12765)
+
+        ONE_MINUTE = 60
+        ONE_HOUR = 60 * ONE_MINUTE
+        ONE_DAY = 24 * ONE_HOUR
+        ONE_YEAR = 365 * ONE_DAY
+
+        self.settings.integer(
+            ['max-age-in-seconds', 'max-age'],
+            'maximum age of a finished job in seconds',
+            metavar='SECONDS',
+            default=ONE_YEAR)
+
+        self.settings.integer(
+            ['debug-now'],
+            'for tests and debugging, '
+            'set current time to SECONDS since the epoch '
+            '(set to 0 to use real time',
+            metavar='SECONDS')
+
+    def process_args(self, args):
+        logging.info('Removing old jobs from Lorry Controller STATEDB')
+
+        job_ids = self.list_jobs()
+        job_infos = self.get_job_infos(job_ids)
+        ids_of_jobs_to_remove = self.select_for_removal(job_infos)
+        self.remove_jobs(ids_of_jobs_to_remove)
+
+    def list_jobs(self):
+        data = self.get('/1.0/list-jobs')
+        obj = json.loads(data)
+        return obj['job_ids']
+
+    def get(self, path):
+        url = self.make_url(path)
+        with contextlib.closing(urllib2.urlopen(url)) as f:
+            return f.read()
+
+    def make_url(self, path):
+        scheme = 'http'
+        netloc = '%s:%s' % (
+            self.settings['webapp-host'], self.settings['webapp-port'])
+        query = None
+        fragment = None
+        parts = (scheme, netloc, path, query, fragment)
+        return urlparse.urlunsplit(parts)
+
+    def get_job_infos(self, job_ids):
+        job_infos = []
+        for job_id in job_ids:
+            try:
+                job_infos.append(self.get_job_info(job_id))
+            except urllib2.HTTPError as e:
+                logging.warning(
+                    'Trouble getting job info for job %s: %s' % 
+                    (job_id, str(e)))
+        return job_infos
+
+    def get_job_info(self, job_id):
+        data = self.get('/1.0/job/%s' % job_id)
+        obj = json.loads(data)
+        exit_code = obj['exit']
+        if obj['job_ended']:
+            exit_timestamp = self.parse_timestamp(obj['job_ended'])
+        else:
+            exit_timestamp = None
+        return JobInfo(job_id, exit_code, exit_timestamp)
+
+    def parse_timestamp(self, timestamp):
+        return time.mktime(time.strptime(timestamp, '%Y-%m-%d %H:%M:%S UTC'))
+
+    def select_for_removal(self, job_infos):
+        return [job_info for job_info in job_infos if self.is_old(job_info)]
+
+    def is_old(self, job_info):
+        if job_info.exit_timestamp is None:
+            return False
+        current_time = self.get_current_time()
+        age_in_seconds = current_time - job_info.exit_timestamp
+        return age_in_seconds >= self.settings['max-age-in-seconds']
+
+    def get_current_time(self):
+        if self.settings['debug-now']:
+            return self.settings['debug-now']
+        return time.time()
+
+    def remove_jobs(self, job_infos):
+        for job_info in job_infos:
+            self.remove_job(job_info.job_id)
+
+    def remove_job(self, job_id):
+        logging.info('Removing job %s', job_id)
+        self.post('/1.0/remove-job', 'job_id=%s' % job_id)
+
+    def post(self, path, data):
+        url = self.make_url(path)
+        f = urllib2.urlopen(url, data)
+        result = f.read()
+        f.close()
+
+
+OldJobRemover().run()