blob: de366f61dc7ad0a23f81349180ba7755b4c528e3 [file] [log] [blame] [edit]
import abc
import datetime
import glob
import logging
import os
import time
import common
from autotest_lib.server import frontend
_AFE = frontend.AFE(debug=False)
JOB_TIME_FORMAT = '%Y-%m-%d %H:%M:%S'
def _is_job_expired(age_limit, timestamp):
"""Check whether a job timestamp is older than an age limit.
@param age_limit: Minimum age, measured in days. If the value is
not positive, the job is always expired.
@param timestamp: Timestamp of the job whose age we are checking.
The format must match JOB_TIME_FORMAT.
@returns True iff the job is old enough to be expired.
"""
if age_limit <= 0:
return True
job_time = datetime.datetime.strptime(timestamp, JOB_TIME_FORMAT)
expiration = job_time + datetime.timedelta(days=age_limit)
return datetime.datetime.now() >= expiration
class _JobDirectory(object):
"""State associated with a job to be offloaded.
The full life-cycle of a job (including failure events that
normally don't occur) looks like this:
1. The job's results directory is discovered by
`get_job_directories()`, and a job instance is created for it.
2. Calls to `offload()` have no effect so long as the job
isn't complete in the database and the job isn't expired
according to the `age_limit` parameter.
3. Eventually, the job is both finished and expired. The next
call to `offload()` makes the first attempt to offload the
directory to GS. Offload is attempted, but fails to complete
(e.g. because of a GS problem).
4. After the first failed offload `is_offloaded()` is false,
but `is_reportable()` is also false, so the failure is not
reported.
5. Another call to `offload()` again tries to offload the
directory, and again fails.
6. After a second failure, `is_offloaded()` is false and
`is_reportable()` is true, so the failure generates an e-mail
notification.
7. Finally, a call to `offload()` succeeds, and the directory no
longer exists. Now `is_offloaded()` is true, so the job
instance is deleted, and future failures will not mention this
directory any more.
Only steps 1. and 7. are guaranteed to occur. The others depend
on the timing of calls to `offload()`, and on the reliability of
the actual offload process.
"""
__metaclass__ = abc.ABCMeta
GLOB_PATTERN = None # must be redefined in subclass
def __init__(self, resultsdir):
self._dirname = resultsdir
self._id = os.path.basename(resultsdir).split('-')[0]
self._offload_count = 0
self._first_offload_start = 0
@classmethod
def get_job_directories(cls):
"""Return a list of directories of jobs that need offloading."""
return [d for d in glob.glob(cls.GLOB_PATTERN) if os.path.isdir(d)]
@abc.abstractmethod
def get_timestamp_if_finished(self):
"""Return this job's timestamp from the database.
If the database has not marked the job as finished, return
`None`. Otherwise, return a timestamp for the job. The
timestamp is to be used to determine expiration in
`_is_job_expired()`.
@return Return `None` if the job is still running; otherwise
return a string with a timestamp in the appropriate
format.
"""
raise NotImplementedError("_JobDirectory.get_timestamp_if_finished")
def enqueue_offload(self, queue, age_limit):
"""Enqueue the job for offload, if it's eligible.
The job is eligible for offloading if the database has marked
it finished, and the job is older than the `age_limit`
parameter.
If the job is eligible, offload processing is requested by
passing the `queue` parameter's `put()` method a sequence with
the job's `_dirname` attribute and its directory name.
@param queue If the job should be offloaded, put the offload
parameters into this queue for processing.
@param age_limit Minimum age for a job to be offloaded. A value
of 0 means that the job will be offloaded as
soon as it is finished.
"""
if not self._offload_count:
timestamp = self.get_timestamp_if_finished()
if not timestamp:
logging.debug('Skipping %s - not finished.', self._dirname)
return
if not _is_job_expired(age_limit, timestamp):
logging.debug('Skipping %s - not old enough.', self._dirname)
return
self._first_offload_start = time.time()
logging.debug('Processing %s', self._dirname)
self._offload_count += 1
queue.put([self._dirname, os.path.dirname(self._dirname)])
def is_offloaded(self):
"""Return whether this job has been successfully offloaded."""
return not os.path.exists(self._dirname)
def is_reportable(self):
"""Return whether this job has a reportable failure."""
return self._offload_count > 1
def get_failure_time(self):
"""Return the time of the first offload failure."""
return self._first_offload_start
def get_failure_count(self):
"""Return the number of times this job has failed to offload."""
return self._offload_count
def get_job_directory(self):
"""Return the name of this job's results directory."""
return self._dirname
class RegularJobDirectory(_JobDirectory):
"""Subclass of _JobDirectory for regular test jobs."""
GLOB_PATTERN = '[0-9]*-*'
def get_timestamp_if_finished(self):
entry = _AFE.run('get_jobs', id=self._id, finished=True)
return entry[0]['created_on'] if entry else None
class SpecialJobDirectory(_JobDirectory):
"""Subclass of _JobDirectory for special (per-host) jobs."""
GLOB_PATTERN = 'hosts/*/[0-9]*-*'
def __init__(self, resultsdir):
super(SpecialJobDirectory, self).__init__(resultsdir)
def get_timestamp_if_finished(self):
entry = _AFE.run('get_special_tasks', id=self._id, is_complete=True)
return entry[0]['time_started'] if entry else None