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.models.job import Job
  33
  34 class Error(Exception):
  35   """Base class for all exceptions raised by this module.
  36   """
  37   pass
  38
  39
  40 class FatalJobError(Error):
  41   """Class for all errors that lead to immediate job abortion.
  42   """
  43   pass
  44
  45
  46 class Handler(object):
  47   """A handler that dispatches a cron job.
  48
  49   The tasks that are mapped into tasks will be called when a worker
  50   has claimed the job. However, there is no guarantee as to how long
  51   the task will be allowed to run. If an Exception is raised the task
  52   is automatically rescheduled for execution.
  53   """
  54
  55   def __init__(self):
  56     """Constructs a new Handler with all known jobs set.
  57     """
  58
  59     self.OUT_OF_TIME = 0
  60     self.ALREADY_CLAIMED = 1
  61     self.SUCCESS = 2
  62     self.ABORTED = 3
  63     self.ERRORED = 4
  64
  65     self.tasks = {}
  66     self.tasks['setupStudentProposalMailing'] = \
  67         student_proposal_mailer.setupStudentProposalMailing
  68     self.tasks['sendStudentProposalMail'] = \
  69         student_proposal_mailer.sendStudentProposalMail
  70
  71   def claimJob(self, job_key):
  72     """A transaction to claim a job.
  73
  74     The transaction is rolled back if the status is not 'waiting'.
  75     """
  76
  77     job = Job.get_by_id(job_key)
  78
  79     if job.status != 'waiting':
  80       raise db.Rollback()
  81
  82     job.status = 'started'
  83
  84     if job.put():
  85       return job
  86     else:
  87       return None
  88
  89   def timeoutJob(self, job):
  90     """Timeout a job.
  91
  92     If a job has timed out more than 50 times, the job is aborted.
  93     """
  94
  95     job.timeouts += 1
  96
  97     if job.timeouts > 50:
  98       job.status = 'aborted'
  99     else:
 100       job.status = 'waiting'
 101
 102     job.put()
 103
 104     job_id = job.key().id()
 105     logging.debug("job %d now timeout %d time(s)" % (job_id, job.timeouts))
 106
 107   def failJob(self, job):
 108     """Fail a job.
 109
 110     If the job has failed more than 5 times, the job is aborted.
 111     """
 112
 113     job.errors += 1
 114
 115     if job.errors > 5:
 116       job.status = 'aborted'
 117     else:
 118       job.status = 'waiting'
 119
 120     job.put()
 121
 122     job_id = job.key().id()
 123     logging.warning("job %d now failed %d time(s)" % (job_id, job.errors))
 124
 125   def finishJob(self, job):
 126     """Finish a job.
 127     """
 128
 129     job.status = 'finished'
 130     job.put()
 131
 132   def abortJob(self, job):
 133     """Abort a job.
 134     """
 135
 136     job.status = 'aborted'
 137     job.put()
 138
 139   def handle(self, job_key):
 140     """Handle one job.
 141
 142     Returns: one of the following status codes:
 143       self.OUT_OF_TIME: returned when a DeadlineExceededError is raised
 144       self.ALREADY_CLAIMED: if job.status is not 'waiting'
 145       self.SUCCESS: if the job.status has been set to 'succes'
 146       self.ABORTED: if the job.status has been set to 'aborted'
 147       self.ERRORED: if the job encountered an error
 148     """
 149
 150     job = None
 151
 152     try:
 153       job = db.run_in_transaction(self.claimJob, job_key)
 154
 155       if not job:
 156         # someone already claimed the job
 157         return self.ALREADY_CLAIMED
 158
 159       if job.task_name not in self.tasks:
 160         logging.error("Unknown job %s" % job.task_name)
 161         db.run_in_transaction(self.abortJob, job_key)
 162         return self.ABORTED
 163
 164       task = self.tasks[job.task_name]
 165
 166       # execute the actual job
 167       task(job)
 168
 169       self.finishJob(job)
 170       return self.SUCCESS
 171     except DeadlineExceededError, exception:
 172       if job:
 173         self.timeoutJob(job)
 174       return self.OUT_OF_TIME
 175     except FatalJobError, exception:
 176       logging.exception(exception)
 177       if job:
 178         self.abortJob(job)
 179       return self.ABORTED
 180     except Exception, exception:
 181       logging.exception(exception)
 182       if job:
 183         self.failJob(job)
 184       return self.ERRORED
 185
 186   def iterate(self, jobs, retry_jobs):
 187     """Trivial iterator that iterates over jobs then retry_jobs
 188     """
 189
 190     for job in jobs:
 191       yield job
 192     while retry_jobs:
 193       yield retry_jobs[0]
 194
 195 handler = Handler()