app/soc/cron/job.py

   1 #!/usr/bin/python2.5
   2 #
   3 # Copyright 2009 the Melange authors.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #   http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 """Cron jobs.
  18 """
  19
  20 __authors__ = [
  21     '"Sverre Rabbelier" <sverre@rabbelier.nl>',
  22     '"Lennard de Rijk" <ljvderijk@gmail.com>',
  23   ]
  24
  25
  26 import logging
  27
  28 from google.appengine.ext import db
  29 from google.appengine.runtime import DeadlineExceededError
  30
  31 from soc.cron import student_proposal_mailer
  32 from soc.cron import unique_user_id_adder
  33 from soc.models.job import Job
  34
  35
  36 class Error(Exception):
  37   """Base class for all exceptions raised by this module.
  38   """
  39   pass
  40
  41
  42 class FatalJobError(Error):
  43   """Class for all errors that lead to immediate job abortion.
  44   """
  45   pass
  46
  47
  48 class Handler(object):
  49   """A handler that dispatches a cron job.
  50
  51   The tasks that are mapped into tasks will be called when a worker
  52   has claimed the job. However, there is no guarantee as to how long
  53   the task will be allowed to run. If an Exception is raised the task
  54   is automatically rescheduled for execution.
  55   """
  56
  57   def __init__(self):
  58     """Constructs a new Handler with all known jobs set.
  59     """
  60     # pylint: disable-msg=C0103
  61     self.OUT_OF_TIME = 0
  62     self.ALREADY_CLAIMED = 1
  63     self.SUCCESS = 2
  64     self.ABORTED = 3
  65     self.ERRORED = 4
  66
  67     self.tasks = {}
  68     self.tasks['setupStudentProposalMailing'] = \
  69         student_proposal_mailer.setupStudentProposalMailing
  70     self.tasks['sendStudentProposalMail'] = \
  71         student_proposal_mailer.sendStudentProposalMail
  72     self.tasks['setupUniqueUserIdAdder'] = \
  73         unique_user_id_adder.setupUniqueUserIdAdder
  74     self.tasks['addUniqueUserIds'] = \
  75         unique_user_id_adder.addUniqueUserIds
  76
  77   def claimJob(self, job_key):
  78     """A transaction to claim a job.
  79
  80     The transaction is rolled back if the status is not 'waiting'.
  81     """
  82
  83     job = Job.get_by_id(job_key)
  84
  85     if job.status != 'waiting':
  86       raise db.Rollback()
  87
  88     job.status = 'started'
  89
  90     # pylint: disable-msg=E1103
  91     if job.put():
  92       return job
  93     else:
  94       return None
  95
  96   def timeoutJob(self, job):
  97     """Timeout a job.
  98
  99     If a job has timed out more than 50 times, the job is aborted.
 100     """
 101
 102     job.timeouts += 1
 103
 104     if job.timeouts > 50:
 105       job.status = 'aborted'
 106     else:
 107       job.status = 'waiting'
 108
 109     job.put()
 110
 111     job_id = job.key().id()
 112     logging.debug("job %d now timeout %d time(s)" % (job_id, job.timeouts))
 113
 114   def failJob(self, job):
 115     """Fail a job.
 116
 117     If the job has failed more than 5 times, the job is aborted.
 118     """
 119
 120     job.errors += 1
 121
 122     if job.errors > 5:
 123       job.status = 'aborted'
 124     else:
 125       job.status = 'waiting'
 126
 127     job.put()
 128
 129     job_id = job.key().id()
 130     logging.warning("job %d now failed %d time(s)" % (job_id, job.errors))
 131
 132   def finishJob(self, job):
 133     """Finish a job.
 134     """
 135
 136     job.status = 'finished'
 137     job.put()
 138
 139   def abortJob(self, job):
 140     """Abort a job.
 141     """
 142
 143     job.status = 'aborted'
 144     job.put()
 145
 146   def handle(self, job_key):
 147     """Handle one job.
 148
 149     Returns: one of the following status codes:
 150       self.OUT_OF_TIME: returned when a DeadlineExceededError is raised
 151       self.ALREADY_CLAIMED: if job.status is not 'waiting'
 152       self.SUCCESS: if the job.status has been set to 'succes'
 153       self.ABORTED: if the job.status has been set to 'aborted'
 154       self.ERRORED: if the job encountered an error
 155     """
 156
 157     job = None
 158
 159     try:
 160       job = db.run_in_transaction(self.claimJob, job_key)
 161
 162       if not job:
 163         # someone already claimed the job
 164         return self.ALREADY_CLAIMED
 165
 166       if job.task_name not in self.tasks:
 167         logging.error("Unknown job %s" % job.task_name)
 168         db.run_in_transaction(self.abortJob, job_key)
 169         return self.ABORTED
 170
 171       task = self.tasks[job.task_name]
 172
 173       # execute the actual job
 174       task(job)
 175
 176       self.finishJob(job)
 177       return self.SUCCESS
 178     except DeadlineExceededError, exception:
 179       if job:
 180         self.timeoutJob(job)
 181       return self.OUT_OF_TIME
 182     except FatalJobError, exception:
 183       logging.exception(exception)
 184       if job:
 185         self.abortJob(job)
 186       return self.ABORTED
 187     except Exception, exception:
 188       logging.exception(exception)
 189       if job:
 190         self.failJob(job)
 191       return self.ERRORED
 192
 193   def iterate(self, jobs, retry_jobs):
 194     """Trivial iterator that iterates over jobs then retry_jobs
 195     """
 196
 197     for job in jobs:
 198       yield job
 199     while retry_jobs:
 200       yield retry_jobs[0]
 201
 202 handler = Handler()