1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
# Copyright 2009-2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""ORM object representing jobs."""
__metaclass__ = type
__all__ = ['InvalidTransition', 'Job', 'JobStatus']
from calendar import timegm
import datetime
import time
import pytz
from sqlobject import (
IntCol,
StringCol,
)
from storm.expr import (
And,
Or,
Select,
)
from zope.interface import implements
from canonical.database.constants import UTC_NOW
from canonical.database.datetimecol import UtcDateTimeCol
from canonical.database.enumcol import EnumCol
from canonical.database.sqlbase import (
quote,
SQLBase,
)
from lp.services.job.interfaces.job import (
IJob,
JobStatus,
LeaseHeld,
)
UTC = pytz.timezone('UTC')
class InvalidTransition(Exception):
"""Invalid transition from one job status to another attempted."""
def __init__(self, current_status, requested_status):
Exception.__init__(
self, 'Transition from %s to %s is invalid.' %
(current_status, requested_status))
class Job(SQLBase):
"""See `IJob`."""
implements(IJob)
scheduled_start = UtcDateTimeCol()
date_created = UtcDateTimeCol()
date_started = UtcDateTimeCol()
date_finished = UtcDateTimeCol()
lease_expires = UtcDateTimeCol()
log = StringCol()
_status = EnumCol(enum=JobStatus, notNull=True, default=JobStatus.WAITING,
dbName='status')
attempt_count = IntCol(default=0)
max_retries = IntCol(default=0)
# Mapping of valid target states from a given state.
_valid_transitions = {
JobStatus.WAITING:
(JobStatus.RUNNING,
JobStatus.SUSPENDED),
JobStatus.RUNNING:
(JobStatus.COMPLETED,
JobStatus.FAILED,
JobStatus.SUSPENDED,
JobStatus.WAITING),
JobStatus.FAILED: (),
JobStatus.COMPLETED: (),
JobStatus.SUSPENDED:
(JobStatus.WAITING,),
}
# Set of all states where the job could eventually complete.
PENDING_STATUSES = frozenset(
(JobStatus.WAITING,
JobStatus.RUNNING,
JobStatus.SUSPENDED))
def _set_status(self, status):
if status not in self._valid_transitions[self._status]:
raise InvalidTransition(self._status, status)
self._status = status
status = property(lambda x: x._status)
@classmethod
def createMultiple(self, store, num_jobs):
"""Create multiple `Job`s at once.
:param store: `Store` to ceate the jobs in.
:param num_jobs: Number of `Job`s to create.
:return: An iterable of `Job.id` values for the new jobs.
"""
job_contents = ["(%s)" % quote(JobStatus.WAITING)] * num_jobs
result = store.execute("""
INSERT INTO Job (status)
VALUES %s
RETURNING id
""" % ", ".join(job_contents))
return [job_id for job_id, in result]
def acquireLease(self, duration=300):
"""See `IJob`."""
if (self.lease_expires is not None
and self.lease_expires >= datetime.datetime.now(UTC)):
raise LeaseHeld
expiry = datetime.datetime.fromtimestamp(time.time() + duration,
UTC)
self.lease_expires = expiry
def getTimeout(self):
"""Return the number of seconds until the job should time out.
Jobs timeout when their leases expire. If the lease for this job has
already expired, return 0.
"""
expiry = timegm(self.lease_expires.timetuple())
return max(0, expiry - time.time())
def start(self):
"""See `IJob`."""
self._set_status(JobStatus.RUNNING)
self.date_started = datetime.datetime.now(UTC)
self.date_finished = None
self.attempt_count += 1
def complete(self):
"""See `IJob`."""
self._set_status(JobStatus.COMPLETED)
self.date_finished = datetime.datetime.now(UTC)
def fail(self):
"""See `IJob`."""
self._set_status(JobStatus.FAILED)
self.date_finished = datetime.datetime.now(UTC)
def queue(self):
"""See `IJob`."""
self._set_status(JobStatus.WAITING)
self.date_finished = datetime.datetime.now(UTC)
def suspend(self):
"""See `IJob`."""
self._set_status(JobStatus.SUSPENDED)
def resume(self):
"""See `IJob`."""
if self.status is not JobStatus.SUSPENDED:
raise InvalidTransition(self._status, JobStatus.WAITING)
self._set_status(JobStatus.WAITING)
Job.ready_jobs = Select(
Job.id,
And(
Job._status == JobStatus.WAITING,
Or(Job.lease_expires == None, Job.lease_expires < UTC_NOW),
Or(Job.scheduled_start == None, Job.scheduled_start <= UTC_NOW),
))
|