Statistics
| Branch: | Tag: | Revision:

root / lib / jqueue.py @ d2c8afb1

History | View | Annotate | Download (41.7 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Module implementing the job queue handling.
23

24
Locking: there's a single, large lock in the L{JobQueue} class. It's
25
used by all other classes in this module.
26

27
@var JOBQUEUE_THREADS: the number of worker threads we start for
28
    processing jobs
29

30
"""
31

    
32
import os
33
import logging
34
import threading
35
import errno
36
import re
37
import time
38
import weakref
39

    
40
from ganeti import constants
41
from ganeti import serializer
42
from ganeti import workerpool
43
from ganeti import opcodes
44
from ganeti import errors
45
from ganeti import mcpu
46
from ganeti import utils
47
from ganeti import jstore
48
from ganeti import rpc
49

    
50

    
51
JOBQUEUE_THREADS = 25
52
JOBS_PER_ARCHIVE_DIRECTORY = 10000
53

    
54

    
55
class CancelJob(Exception):
56
  """Special exception to cancel a job.
57

58
  """
59

    
60

    
61
def TimeStampNow():
62
  """Returns the current timestamp.
63

64
  @rtype: tuple
65
  @return: the current time in the (seconds, microseconds) format
66

67
  """
68
  return utils.SplitTime(time.time())
69

    
70

    
71
class _QueuedOpCode(object):
72
  """Encapsulates an opcode object.
73

74
  @ivar log: holds the execution log and consists of tuples
75
  of the form C{(log_serial, timestamp, level, message)}
76
  @ivar input: the OpCode we encapsulate
77
  @ivar status: the current status
78
  @ivar result: the result of the LU execution
79
  @ivar start_timestamp: timestamp for the start of the execution
80
  @ivar stop_timestamp: timestamp for the end of the execution
81

82
  """
83
  __slots__ = ["input", "status", "result", "log",
84
               "start_timestamp", "end_timestamp",
85
               "__weakref__"]
86

    
87
  def __init__(self, op):
88
    """Constructor for the _QuededOpCode.
89

90
    @type op: L{opcodes.OpCode}
91
    @param op: the opcode we encapsulate
92

93
    """
94
    self.input = op
95
    self.status = constants.OP_STATUS_QUEUED
96
    self.result = None
97
    self.log = []
98
    self.start_timestamp = None
99
    self.end_timestamp = None
100

    
101
  @classmethod
102
  def Restore(cls, state):
103
    """Restore the _QueuedOpCode from the serialized form.
104

105
    @type state: dict
106
    @param state: the serialized state
107
    @rtype: _QueuedOpCode
108
    @return: a new _QueuedOpCode instance
109

110
    """
111
    obj = _QueuedOpCode.__new__(cls)
112
    obj.input = opcodes.OpCode.LoadOpCode(state["input"])
113
    obj.status = state["status"]
114
    obj.result = state["result"]
115
    obj.log = state["log"]
116
    obj.start_timestamp = state.get("start_timestamp", None)
117
    obj.end_timestamp = state.get("end_timestamp", None)
118
    return obj
119

    
120
  def Serialize(self):
121
    """Serializes this _QueuedOpCode.
122

123
    @rtype: dict
124
    @return: the dictionary holding the serialized state
125

126
    """
127
    return {
128
      "input": self.input.__getstate__(),
129
      "status": self.status,
130
      "result": self.result,
131
      "log": self.log,
132
      "start_timestamp": self.start_timestamp,
133
      "end_timestamp": self.end_timestamp,
134
      }
135

    
136

    
137
class _QueuedJob(object):
138
  """In-memory job representation.
139

140
  This is what we use to track the user-submitted jobs. Locking must
141
  be taken care of by users of this class.
142

143
  @type queue: L{JobQueue}
144
  @ivar queue: the parent queue
145
  @ivar id: the job ID
146
  @type ops: list
147
  @ivar ops: the list of _QueuedOpCode that constitute the job
148
  @type log_serial: int
149
  @ivar log_serial: holds the index for the next log entry
150
  @ivar received_timestamp: the timestamp for when the job was received
151
  @ivar start_timestmap: the timestamp for start of execution
152
  @ivar end_timestamp: the timestamp for end of execution
153
  @ivar lock_status: In-memory locking information for debugging
154
  @ivar change: a Condition variable we use for waiting for job changes
155

156
  """
157
  # pylint: disable-msg=W0212
158
  __slots__ = ["queue", "id", "ops", "log_serial",
159
               "received_timestamp", "start_timestamp", "end_timestamp",
160
               "lock_status", "change",
161
               "__weakref__"]
162

    
163
  def __init__(self, queue, job_id, ops):
164
    """Constructor for the _QueuedJob.
165

166
    @type queue: L{JobQueue}
167
    @param queue: our parent queue
168
    @type job_id: job_id
169
    @param job_id: our job id
170
    @type ops: list
171
    @param ops: the list of opcodes we hold, which will be encapsulated
172
        in _QueuedOpCodes
173

174
    """
175
    if not ops:
176
      # TODO: use a better exception
177
      raise Exception("No opcodes")
178

    
179
    self.queue = queue
180
    self.id = job_id
181
    self.ops = [_QueuedOpCode(op) for op in ops]
182
    self.log_serial = 0
183
    self.received_timestamp = TimeStampNow()
184
    self.start_timestamp = None
185
    self.end_timestamp = None
186

    
187
    # In-memory attributes
188
    self.lock_status = None
189

    
190
    # Condition to wait for changes
191
    self.change = threading.Condition(self.queue._lock)
192

    
193
  def __repr__(self):
194
    status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
195
              "id=%s" % self.id,
196
              "ops=%s" % ",".join([op.input.Summary() for op in self.ops])]
197

    
198
    return "<%s at %#x>" % (" ".join(status), id(self))
199

    
200
  @classmethod
201
  def Restore(cls, queue, state):
202
    """Restore a _QueuedJob from serialized state:
203

204
    @type queue: L{JobQueue}
205
    @param queue: to which queue the restored job belongs
206
    @type state: dict
207
    @param state: the serialized state
208
    @rtype: _JobQueue
209
    @return: the restored _JobQueue instance
210

211
    """
212
    obj = _QueuedJob.__new__(cls)
213
    obj.queue = queue
214
    obj.id = state["id"]
215
    obj.received_timestamp = state.get("received_timestamp", None)
216
    obj.start_timestamp = state.get("start_timestamp", None)
217
    obj.end_timestamp = state.get("end_timestamp", None)
218

    
219
    # In-memory attributes
220
    obj.lock_status = None
221

    
222
    obj.ops = []
223
    obj.log_serial = 0
224
    for op_state in state["ops"]:
225
      op = _QueuedOpCode.Restore(op_state)
226
      for log_entry in op.log:
227
        obj.log_serial = max(obj.log_serial, log_entry[0])
228
      obj.ops.append(op)
229

    
230
    # Condition to wait for changes
231
    obj.change = threading.Condition(obj.queue._lock)
232

    
233
    return obj
234

    
235
  def Serialize(self):
236
    """Serialize the _JobQueue instance.
237

238
    @rtype: dict
239
    @return: the serialized state
240

241
    """
242
    return {
243
      "id": self.id,
244
      "ops": [op.Serialize() for op in self.ops],
245
      "start_timestamp": self.start_timestamp,
246
      "end_timestamp": self.end_timestamp,
247
      "received_timestamp": self.received_timestamp,
248
      }
249

    
250
  def CalcStatus(self):
251
    """Compute the status of this job.
252

253
    This function iterates over all the _QueuedOpCodes in the job and
254
    based on their status, computes the job status.
255

256
    The algorithm is:
257
      - if we find a cancelled, or finished with error, the job
258
        status will be the same
259
      - otherwise, the last opcode with the status one of:
260
          - waitlock
261
          - canceling
262
          - running
263

264
        will determine the job status
265

266
      - otherwise, it means either all opcodes are queued, or success,
267
        and the job status will be the same
268

269
    @return: the job status
270

271
    """
272
    status = constants.JOB_STATUS_QUEUED
273

    
274
    all_success = True
275
    for op in self.ops:
276
      if op.status == constants.OP_STATUS_SUCCESS:
277
        continue
278

    
279
      all_success = False
280

    
281
      if op.status == constants.OP_STATUS_QUEUED:
282
        pass
283
      elif op.status == constants.OP_STATUS_WAITLOCK:
284
        status = constants.JOB_STATUS_WAITLOCK
285
      elif op.status == constants.OP_STATUS_RUNNING:
286
        status = constants.JOB_STATUS_RUNNING
287
      elif op.status == constants.OP_STATUS_CANCELING:
288
        status = constants.JOB_STATUS_CANCELING
289
        break
290
      elif op.status == constants.OP_STATUS_ERROR:
291
        status = constants.JOB_STATUS_ERROR
292
        # The whole job fails if one opcode failed
293
        break
294
      elif op.status == constants.OP_STATUS_CANCELED:
295
        status = constants.OP_STATUS_CANCELED
296
        break
297

    
298
    if all_success:
299
      status = constants.JOB_STATUS_SUCCESS
300

    
301
    return status
302

    
303
  def GetLogEntries(self, newer_than):
304
    """Selectively returns the log entries.
305

306
    @type newer_than: None or int
307
    @param newer_than: if this is None, return all log entries,
308
        otherwise return only the log entries with serial higher
309
        than this value
310
    @rtype: list
311
    @return: the list of the log entries selected
312

313
    """
314
    if newer_than is None:
315
      serial = -1
316
    else:
317
      serial = newer_than
318

    
319
    entries = []
320
    for op in self.ops:
321
      entries.extend(filter(lambda entry: entry[0] > serial, op.log))
322

    
323
    return entries
324

    
325
  def MarkUnfinishedOps(self, status, result):
326
    """Mark unfinished opcodes with a given status and result.
327

328
    This is an utility function for marking all running or waiting to
329
    be run opcodes with a given status. Opcodes which are already
330
    finalised are not changed.
331

332
    @param status: a given opcode status
333
    @param result: the opcode result
334

335
    """
336
    not_marked = True
337
    for op in self.ops:
338
      if op.status in constants.OPS_FINALIZED:
339
        assert not_marked, "Finalized opcodes found after non-finalized ones"
340
        continue
341
      op.status = status
342
      op.result = result
343
      not_marked = False
344

    
345

    
346
class _OpExecCallbacks(mcpu.OpExecCbBase):
347
  def __init__(self, queue, job, op):
348
    """Initializes this class.
349

350
    @type queue: L{JobQueue}
351
    @param queue: Job queue
352
    @type job: L{_QueuedJob}
353
    @param job: Job object
354
    @type op: L{_QueuedOpCode}
355
    @param op: OpCode
356

357
    """
358
    assert queue, "Queue is missing"
359
    assert job, "Job is missing"
360
    assert op, "Opcode is missing"
361

    
362
    self._queue = queue
363
    self._job = job
364
    self._op = op
365

    
366
  def NotifyStart(self):
367
    """Mark the opcode as running, not lock-waiting.
368

369
    This is called from the mcpu code as a notifier function, when the LU is
370
    finally about to start the Exec() method. Of course, to have end-user
371
    visible results, the opcode must be initially (before calling into
372
    Processor.ExecOpCode) set to OP_STATUS_WAITLOCK.
373

374
    """
375
    self._queue.acquire()
376
    try:
377
      assert self._op.status in (constants.OP_STATUS_WAITLOCK,
378
                                 constants.OP_STATUS_CANCELING)
379

    
380
      # All locks are acquired by now
381
      self._job.lock_status = None
382

    
383
      # Cancel here if we were asked to
384
      if self._op.status == constants.OP_STATUS_CANCELING:
385
        raise CancelJob()
386

    
387
      self._op.status = constants.OP_STATUS_RUNNING
388
    finally:
389
      self._queue.release()
390

    
391
  def Feedback(self, *args):
392
    """Append a log entry.
393

394
    """
395
    assert len(args) < 3
396

    
397
    if len(args) == 1:
398
      log_type = constants.ELOG_MESSAGE
399
      log_msg = args[0]
400
    else:
401
      (log_type, log_msg) = args
402

    
403
    # The time is split to make serialization easier and not lose
404
    # precision.
405
    timestamp = utils.SplitTime(time.time())
406

    
407
    self._queue.acquire()
408
    try:
409
      self._job.log_serial += 1
410
      self._op.log.append((self._job.log_serial, timestamp, log_type, log_msg))
411

    
412
      self._job.change.notifyAll()
413
    finally:
414
      self._queue.release()
415

    
416
  def ReportLocks(self, msg):
417
    """Write locking information to the job.
418

419
    Called whenever the LU processor is waiting for a lock or has acquired one.
420

421
    """
422
    # Not getting the queue lock because this is a single assignment
423
    self._job.lock_status = msg
424

    
425

    
426
class _JobQueueWorker(workerpool.BaseWorker):
427
  """The actual job workers.
428

429
  """
430
  def RunTask(self, job): # pylint: disable-msg=W0221
431
    """Job executor.
432

433
    This functions processes a job. It is closely tied to the _QueuedJob and
434
    _QueuedOpCode classes.
435

436
    @type job: L{_QueuedJob}
437
    @param job: the job to be processed
438

439
    """
440
    logging.info("Processing job %s", job.id)
441
    proc = mcpu.Processor(self.pool.queue.context, job.id)
442
    queue = job.queue
443
    try:
444
      try:
445
        count = len(job.ops)
446
        for idx, op in enumerate(job.ops):
447
          op_summary = op.input.Summary()
448
          if op.status == constants.OP_STATUS_SUCCESS:
449
            # this is a job that was partially completed before master
450
            # daemon shutdown, so it can be expected that some opcodes
451
            # are already completed successfully (if any did error
452
            # out, then the whole job should have been aborted and not
453
            # resubmitted for processing)
454
            logging.info("Op %s/%s: opcode %s already processed, skipping",
455
                         idx + 1, count, op_summary)
456
            continue
457
          try:
458
            logging.info("Op %s/%s: Starting opcode %s", idx + 1, count,
459
                         op_summary)
460

    
461
            queue.acquire()
462
            try:
463
              if op.status == constants.OP_STATUS_CANCELED:
464
                raise CancelJob()
465
              assert op.status == constants.OP_STATUS_QUEUED
466
              op.status = constants.OP_STATUS_WAITLOCK
467
              op.result = None
468
              op.start_timestamp = TimeStampNow()
469
              if idx == 0: # first opcode
470
                job.start_timestamp = op.start_timestamp
471
              queue.UpdateJobUnlocked(job)
472

    
473
              input_opcode = op.input
474
            finally:
475
              queue.release()
476

    
477
            # Make sure not to hold queue lock while calling ExecOpCode
478
            result = proc.ExecOpCode(input_opcode,
479
                                     _OpExecCallbacks(queue, job, op))
480

    
481
            queue.acquire()
482
            try:
483
              op.status = constants.OP_STATUS_SUCCESS
484
              op.result = result
485
              op.end_timestamp = TimeStampNow()
486
              queue.UpdateJobUnlocked(job)
487
            finally:
488
              queue.release()
489

    
490
            logging.info("Op %s/%s: Successfully finished opcode %s",
491
                         idx + 1, count, op_summary)
492
          except CancelJob:
493
            # Will be handled further up
494
            raise
495
          except Exception, err:
496
            queue.acquire()
497
            try:
498
              try:
499
                op.status = constants.OP_STATUS_ERROR
500
                if isinstance(err, errors.GenericError):
501
                  op.result = errors.EncodeException(err)
502
                else:
503
                  op.result = str(err)
504
                op.end_timestamp = TimeStampNow()
505
                logging.info("Op %s/%s: Error in opcode %s: %s",
506
                             idx + 1, count, op_summary, err)
507
              finally:
508
                queue.UpdateJobUnlocked(job)
509
            finally:
510
              queue.release()
511
            raise
512

    
513
      except CancelJob:
514
        queue.acquire()
515
        try:
516
          queue.CancelJobUnlocked(job)
517
        finally:
518
          queue.release()
519
      except errors.GenericError, err:
520
        logging.exception("Ganeti exception")
521
      except:
522
        logging.exception("Unhandled exception")
523
    finally:
524
      queue.acquire()
525
      try:
526
        try:
527
          job.lock_status = None
528
          job.end_timestamp = TimeStampNow()
529
          queue.UpdateJobUnlocked(job)
530
        finally:
531
          job_id = job.id
532
          status = job.CalcStatus()
533
      finally:
534
        queue.release()
535

    
536
      logging.info("Finished job %s, status = %s", job_id, status)
537

    
538

    
539
class _JobQueueWorkerPool(workerpool.WorkerPool):
540
  """Simple class implementing a job-processing workerpool.
541

542
  """
543
  def __init__(self, queue):
544
    super(_JobQueueWorkerPool, self).__init__("JobQueue",
545
                                              JOBQUEUE_THREADS,
546
                                              _JobQueueWorker)
547
    self.queue = queue
548

    
549

    
550
def _RequireOpenQueue(fn):
551
  """Decorator for "public" functions.
552

553
  This function should be used for all 'public' functions. That is,
554
  functions usually called from other classes. Note that this should
555
  be applied only to methods (not plain functions), since it expects
556
  that the decorated function is called with a first argument that has
557
  a '_queue_lock' argument.
558

559
  @warning: Use this decorator only after utils.LockedMethod!
560

561
  Example::
562
    @utils.LockedMethod
563
    @_RequireOpenQueue
564
    def Example(self):
565
      pass
566

567
  """
568
  def wrapper(self, *args, **kwargs):
569
    # pylint: disable-msg=W0212
570
    assert self._queue_lock is not None, "Queue should be open"
571
    return fn(self, *args, **kwargs)
572
  return wrapper
573

    
574

    
575
class JobQueue(object):
576
  """Queue used to manage the jobs.
577

578
  @cvar _RE_JOB_FILE: regex matching the valid job file names
579

580
  """
581
  _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
582

    
583
  def __init__(self, context):
584
    """Constructor for JobQueue.
585

586
    The constructor will initialize the job queue object and then
587
    start loading the current jobs from disk, either for starting them
588
    (if they were queue) or for aborting them (if they were already
589
    running).
590

591
    @type context: GanetiContext
592
    @param context: the context object for access to the configuration
593
        data and other ganeti objects
594

595
    """
596
    self.context = context
597
    self._memcache = weakref.WeakValueDictionary()
598
    self._my_hostname = utils.HostInfo().name
599

    
600
    # Locking
601
    self._lock = threading.Lock()
602
    self.acquire = self._lock.acquire
603
    self.release = self._lock.release
604

    
605
    # Initialize
606
    self._queue_lock = jstore.InitAndVerifyQueue(must_lock=True)
607

    
608
    # Read serial file
609
    self._last_serial = jstore.ReadSerial()
610
    assert self._last_serial is not None, ("Serial file was modified between"
611
                                           " check in jstore and here")
612

    
613
    # Get initial list of nodes
614
    self._nodes = dict((n.name, n.primary_ip)
615
                       for n in self.context.cfg.GetAllNodesInfo().values()
616
                       if n.master_candidate)
617

    
618
    # Remove master node
619
    try:
620
      del self._nodes[self._my_hostname]
621
    except KeyError:
622
      pass
623

    
624
    # TODO: Check consistency across nodes
625

    
626
    # Setup worker pool
627
    self._wpool = _JobQueueWorkerPool(self)
628
    try:
629
      # We need to lock here because WorkerPool.AddTask() may start a job while
630
      # we're still doing our work.
631
      self.acquire()
632
      try:
633
        logging.info("Inspecting job queue")
634

    
635
        all_job_ids = self._GetJobIDsUnlocked()
636
        jobs_count = len(all_job_ids)
637
        lastinfo = time.time()
638
        for idx, job_id in enumerate(all_job_ids):
639
          # Give an update every 1000 jobs or 10 seconds
640
          if (idx % 1000 == 0 or time.time() >= (lastinfo + 10.0) or
641
              idx == (jobs_count - 1)):
642
            logging.info("Job queue inspection: %d/%d (%0.1f %%)",
643
                         idx, jobs_count - 1, 100.0 * (idx + 1) / jobs_count)
644
            lastinfo = time.time()
645

    
646
          job = self._LoadJobUnlocked(job_id)
647

    
648
          # a failure in loading the job can cause 'None' to be returned
649
          if job is None:
650
            continue
651

    
652
          status = job.CalcStatus()
653

    
654
          if status in (constants.JOB_STATUS_QUEUED, ):
655
            self._wpool.AddTask(job)
656

    
657
          elif status in (constants.JOB_STATUS_RUNNING,
658
                          constants.JOB_STATUS_WAITLOCK,
659
                          constants.JOB_STATUS_CANCELING):
660
            logging.warning("Unfinished job %s found: %s", job.id, job)
661
            try:
662
              job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,
663
                                    "Unclean master daemon shutdown")
664
            finally:
665
              self.UpdateJobUnlocked(job)
666

    
667
        logging.info("Job queue inspection finished")
668
      finally:
669
        self.release()
670
    except:
671
      self._wpool.TerminateWorkers()
672
      raise
673

    
674
  @utils.LockedMethod
675
  @_RequireOpenQueue
676
  def AddNode(self, node):
677
    """Register a new node with the queue.
678

679
    @type node: L{objects.Node}
680
    @param node: the node object to be added
681

682
    """
683
    node_name = node.name
684
    assert node_name != self._my_hostname
685

    
686
    # Clean queue directory on added node
687
    result = rpc.RpcRunner.call_jobqueue_purge(node_name)
688
    msg = result.fail_msg
689
    if msg:
690
      logging.warning("Cannot cleanup queue directory on node %s: %s",
691
                      node_name, msg)
692

    
693
    if not node.master_candidate:
694
      # remove if existing, ignoring errors
695
      self._nodes.pop(node_name, None)
696
      # and skip the replication of the job ids
697
      return
698

    
699
    # Upload the whole queue excluding archived jobs
700
    files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
701

    
702
    # Upload current serial file
703
    files.append(constants.JOB_QUEUE_SERIAL_FILE)
704

    
705
    for file_name in files:
706
      # Read file content
707
      content = utils.ReadFile(file_name)
708

    
709
      result = rpc.RpcRunner.call_jobqueue_update([node_name],
710
                                                  [node.primary_ip],
711
                                                  file_name, content)
712
      msg = result[node_name].fail_msg
713
      if msg:
714
        logging.error("Failed to upload file %s to node %s: %s",
715
                      file_name, node_name, msg)
716

    
717
    self._nodes[node_name] = node.primary_ip
718

    
719
  @utils.LockedMethod
720
  @_RequireOpenQueue
721
  def RemoveNode(self, node_name):
722
    """Callback called when removing nodes from the cluster.
723

724
    @type node_name: str
725
    @param node_name: the name of the node to remove
726

727
    """
728
    try:
729
      # The queue is removed by the "leave node" RPC call.
730
      del self._nodes[node_name]
731
    except KeyError:
732
      pass
733

    
734
  @staticmethod
735
  def _CheckRpcResult(result, nodes, failmsg):
736
    """Verifies the status of an RPC call.
737

738
    Since we aim to keep consistency should this node (the current
739
    master) fail, we will log errors if our rpc fail, and especially
740
    log the case when more than half of the nodes fails.
741

742
    @param result: the data as returned from the rpc call
743
    @type nodes: list
744
    @param nodes: the list of nodes we made the call to
745
    @type failmsg: str
746
    @param failmsg: the identifier to be used for logging
747

748
    """
749
    failed = []
750
    success = []
751

    
752
    for node in nodes:
753
      msg = result[node].fail_msg
754
      if msg:
755
        failed.append(node)
756
        logging.error("RPC call %s (%s) failed on node %s: %s",
757
                      result[node].call, failmsg, node, msg)
758
      else:
759
        success.append(node)
760

    
761
    # +1 for the master node
762
    if (len(success) + 1) < len(failed):
763
      # TODO: Handle failing nodes
764
      logging.error("More than half of the nodes failed")
765

    
766
  def _GetNodeIp(self):
767
    """Helper for returning the node name/ip list.
768

769
    @rtype: (list, list)
770
    @return: a tuple of two lists, the first one with the node
771
        names and the second one with the node addresses
772

773
    """
774
    name_list = self._nodes.keys()
775
    addr_list = [self._nodes[name] for name in name_list]
776
    return name_list, addr_list
777

    
778
  def _WriteAndReplicateFileUnlocked(self, file_name, data):
779
    """Writes a file locally and then replicates it to all nodes.
780

781
    This function will replace the contents of a file on the local
782
    node and then replicate it to all the other nodes we have.
783

784
    @type file_name: str
785
    @param file_name: the path of the file to be replicated
786
    @type data: str
787
    @param data: the new contents of the file
788

789
    """
790
    utils.WriteFile(file_name, data=data)
791

    
792
    names, addrs = self._GetNodeIp()
793
    result = rpc.RpcRunner.call_jobqueue_update(names, addrs, file_name, data)
794
    self._CheckRpcResult(result, self._nodes,
795
                         "Updating %s" % file_name)
796

    
797
  def _RenameFilesUnlocked(self, rename):
798
    """Renames a file locally and then replicate the change.
799

800
    This function will rename a file in the local queue directory
801
    and then replicate this rename to all the other nodes we have.
802

803
    @type rename: list of (old, new)
804
    @param rename: List containing tuples mapping old to new names
805

806
    """
807
    # Rename them locally
808
    for old, new in rename:
809
      utils.RenameFile(old, new, mkdir=True)
810

    
811
    # ... and on all nodes
812
    names, addrs = self._GetNodeIp()
813
    result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
814
    self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
815

    
816
  @staticmethod
817
  def _FormatJobID(job_id):
818
    """Convert a job ID to string format.
819

820
    Currently this just does C{str(job_id)} after performing some
821
    checks, but if we want to change the job id format this will
822
    abstract this change.
823

824
    @type job_id: int or long
825
    @param job_id: the numeric job id
826
    @rtype: str
827
    @return: the formatted job id
828

829
    """
830
    if not isinstance(job_id, (int, long)):
831
      raise errors.ProgrammerError("Job ID '%s' not numeric" % job_id)
832
    if job_id < 0:
833
      raise errors.ProgrammerError("Job ID %s is negative" % job_id)
834

    
835
    return str(job_id)
836

    
837
  @classmethod
838
  def _GetArchiveDirectory(cls, job_id):
839
    """Returns the archive directory for a job.
840

841
    @type job_id: str
842
    @param job_id: Job identifier
843
    @rtype: str
844
    @return: Directory name
845

846
    """
847
    return str(int(job_id) / JOBS_PER_ARCHIVE_DIRECTORY)
848

    
849
  def _NewSerialsUnlocked(self, count):
850
    """Generates a new job identifier.
851

852
    Job identifiers are unique during the lifetime of a cluster.
853

854
    @type count: integer
855
    @param count: how many serials to return
856
    @rtype: str
857
    @return: a string representing the job identifier.
858

859
    """
860
    assert count > 0
861
    # New number
862
    serial = self._last_serial + count
863

    
864
    # Write to file
865
    self._WriteAndReplicateFileUnlocked(constants.JOB_QUEUE_SERIAL_FILE,
866
                                        "%s\n" % serial)
867

    
868
    result = [self._FormatJobID(v)
869
              for v in range(self._last_serial, serial + 1)]
870
    # Keep it only if we were able to write the file
871
    self._last_serial = serial
872

    
873
    return result
874

    
875
  @staticmethod
876
  def _GetJobPath(job_id):
877
    """Returns the job file for a given job id.
878

879
    @type job_id: str
880
    @param job_id: the job identifier
881
    @rtype: str
882
    @return: the path to the job file
883

884
    """
885
    return os.path.join(constants.QUEUE_DIR, "job-%s" % job_id)
886

    
887
  @classmethod
888
  def _GetArchivedJobPath(cls, job_id):
889
    """Returns the archived job file for a give job id.
890

891
    @type job_id: str
892
    @param job_id: the job identifier
893
    @rtype: str
894
    @return: the path to the archived job file
895

896
    """
897
    path = "%s/job-%s" % (cls._GetArchiveDirectory(job_id), job_id)
898
    return os.path.join(constants.JOB_QUEUE_ARCHIVE_DIR, path)
899

    
900
  @classmethod
901
  def _ExtractJobID(cls, name):
902
    """Extract the job id from a filename.
903

904
    @type name: str
905
    @param name: the job filename
906
    @rtype: job id or None
907
    @return: the job id corresponding to the given filename,
908
        or None if the filename does not represent a valid
909
        job file
910

911
    """
912
    m = cls._RE_JOB_FILE.match(name)
913
    if m:
914
      return m.group(1)
915
    else:
916
      return None
917

    
918
  def _GetJobIDsUnlocked(self, archived=False):
919
    """Return all known job IDs.
920

921
    If the parameter archived is True, archived jobs IDs will be
922
    included. Currently this argument is unused.
923

924
    The method only looks at disk because it's a requirement that all
925
    jobs are present on disk (so in the _memcache we don't have any
926
    extra IDs).
927

928
    @rtype: list
929
    @return: the list of job IDs
930

931
    """
932
    # pylint: disable-msg=W0613
933
    jlist = [self._ExtractJobID(name) for name in self._ListJobFiles()]
934
    jlist = utils.NiceSort(jlist)
935
    return jlist
936

    
937
  def _ListJobFiles(self):
938
    """Returns the list of current job files.
939

940
    @rtype: list
941
    @return: the list of job file names
942

943
    """
944
    return [name for name in utils.ListVisibleFiles(constants.QUEUE_DIR)
945
            if self._RE_JOB_FILE.match(name)]
946

    
947
  def _LoadJobUnlocked(self, job_id):
948
    """Loads a job from the disk or memory.
949

950
    Given a job id, this will return the cached job object if
951
    existing, or try to load the job from the disk. If loading from
952
    disk, it will also add the job to the cache.
953

954
    @param job_id: the job id
955
    @rtype: L{_QueuedJob} or None
956
    @return: either None or the job object
957

958
    """
959
    job = self._memcache.get(job_id, None)
960
    if job:
961
      logging.debug("Found job %s in memcache", job_id)
962
      return job
963

    
964
    filepath = self._GetJobPath(job_id)
965
    logging.debug("Loading job from %s", filepath)
966
    try:
967
      raw_data = utils.ReadFile(filepath)
968
    except IOError, err:
969
      if err.errno in (errno.ENOENT, ):
970
        return None
971
      raise
972

    
973
    data = serializer.LoadJson(raw_data)
974

    
975
    try:
976
      job = _QueuedJob.Restore(self, data)
977
    except Exception, err: # pylint: disable-msg=W0703
978
      new_path = self._GetArchivedJobPath(job_id)
979
      if filepath == new_path:
980
        # job already archived (future case)
981
        logging.exception("Can't parse job %s", job_id)
982
      else:
983
        # non-archived case
984
        logging.exception("Can't parse job %s, will archive.", job_id)
985
        self._RenameFilesUnlocked([(filepath, new_path)])
986
      return None
987

    
988
    self._memcache[job_id] = job
989
    logging.debug("Added job %s to the cache", job_id)
990
    return job
991

    
992
  def _GetJobsUnlocked(self, job_ids):
993
    """Return a list of jobs based on their IDs.
994

995
    @type job_ids: list
996
    @param job_ids: either an empty list (meaning all jobs),
997
        or a list of job IDs
998
    @rtype: list
999
    @return: the list of job objects
1000

1001
    """
1002
    if not job_ids:
1003
      job_ids = self._GetJobIDsUnlocked()
1004

    
1005
    return [self._LoadJobUnlocked(job_id) for job_id in job_ids]
1006

    
1007
  @staticmethod
1008
  def _IsQueueMarkedDrain():
1009
    """Check if the queue is marked from drain.
1010

1011
    This currently uses the queue drain file, which makes it a
1012
    per-node flag. In the future this can be moved to the config file.
1013

1014
    @rtype: boolean
1015
    @return: True of the job queue is marked for draining
1016

1017
    """
1018
    return os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
1019

    
1020
  @staticmethod
1021
  def SetDrainFlag(drain_flag):
1022
    """Sets the drain flag for the queue.
1023

1024
    This is similar to the function L{backend.JobQueueSetDrainFlag},
1025
    and in the future we might merge them.
1026

1027
    @type drain_flag: boolean
1028
    @param drain_flag: Whether to set or unset the drain flag
1029

1030
    """
1031
    if drain_flag:
1032
      utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
1033
    else:
1034
      utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
1035
    return True
1036

    
1037
  @_RequireOpenQueue
1038
  def _SubmitJobUnlocked(self, job_id, ops):
1039
    """Create and store a new job.
1040

1041
    This enters the job into our job queue and also puts it on the new
1042
    queue, in order for it to be picked up by the queue processors.
1043

1044
    @type job_id: job ID
1045
    @param job_id: the job ID for the new job
1046
    @type ops: list
1047
    @param ops: The list of OpCodes that will become the new job.
1048
    @rtype: job ID
1049
    @return: the job ID of the newly created job
1050
    @raise errors.JobQueueDrainError: if the job is marked for draining
1051

1052
    """
1053
    if self._IsQueueMarkedDrain():
1054
      raise errors.JobQueueDrainError("Job queue is drained, refusing job")
1055

    
1056
    # Check job queue size
1057
    size = len(self._ListJobFiles())
1058
    if size >= constants.JOB_QUEUE_SIZE_SOFT_LIMIT:
1059
      # TODO: Autoarchive jobs. Make sure it's not done on every job
1060
      # submission, though.
1061
      #size = ...
1062
      pass
1063

    
1064
    if size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
1065
      raise errors.JobQueueFull()
1066

    
1067
    job = _QueuedJob(self, job_id, ops)
1068

    
1069
    # Write to disk
1070
    self.UpdateJobUnlocked(job)
1071

    
1072
    logging.debug("Adding new job %s to the cache", job_id)
1073
    self._memcache[job_id] = job
1074

    
1075
    # Add to worker pool
1076
    self._wpool.AddTask(job)
1077

    
1078
    return job.id
1079

    
1080
  @utils.LockedMethod
1081
  @_RequireOpenQueue
1082
  def SubmitJob(self, ops):
1083
    """Create and store a new job.
1084

1085
    @see: L{_SubmitJobUnlocked}
1086

1087
    """
1088
    job_id = self._NewSerialsUnlocked(1)[0]
1089
    return self._SubmitJobUnlocked(job_id, ops)
1090

    
1091
  @utils.LockedMethod
1092
  @_RequireOpenQueue
1093
  def SubmitManyJobs(self, jobs):
1094
    """Create and store multiple jobs.
1095

1096
    @see: L{_SubmitJobUnlocked}
1097

1098
    """
1099
    results = []
1100
    all_job_ids = self._NewSerialsUnlocked(len(jobs))
1101
    for job_id, ops in zip(all_job_ids, jobs):
1102
      try:
1103
        data = self._SubmitJobUnlocked(job_id, ops)
1104
        status = True
1105
      except errors.GenericError, err:
1106
        data = str(err)
1107
        status = False
1108
      results.append((status, data))
1109

    
1110
    return results
1111

    
1112
  @_RequireOpenQueue
1113
  def UpdateJobUnlocked(self, job):
1114
    """Update a job's on disk storage.
1115

1116
    After a job has been modified, this function needs to be called in
1117
    order to write the changes to disk and replicate them to the other
1118
    nodes.
1119

1120
    @type job: L{_QueuedJob}
1121
    @param job: the changed job
1122

1123
    """
1124
    filename = self._GetJobPath(job.id)
1125
    data = serializer.DumpJson(job.Serialize(), indent=False)
1126
    logging.debug("Writing job %s to %s", job.id, filename)
1127
    self._WriteAndReplicateFileUnlocked(filename, data)
1128

    
1129
    # Notify waiters about potential changes
1130
    job.change.notifyAll()
1131

    
1132
  @utils.LockedMethod
1133
  @_RequireOpenQueue
1134
  def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial,
1135
                        timeout):
1136
    """Waits for changes in a job.
1137

1138
    @type job_id: string
1139
    @param job_id: Job identifier
1140
    @type fields: list of strings
1141
    @param fields: Which fields to check for changes
1142
    @type prev_job_info: list or None
1143
    @param prev_job_info: Last job information returned
1144
    @type prev_log_serial: int
1145
    @param prev_log_serial: Last job message serial number
1146
    @type timeout: float
1147
    @param timeout: maximum time to wait
1148
    @rtype: tuple (job info, log entries)
1149
    @return: a tuple of the job information as required via
1150
        the fields parameter, and the log entries as a list
1151

1152
        if the job has not changed and the timeout has expired,
1153
        we instead return a special value,
1154
        L{constants.JOB_NOTCHANGED}, which should be interpreted
1155
        as such by the clients
1156

1157
    """
1158
    job = self._LoadJobUnlocked(job_id)
1159
    if not job:
1160
      logging.debug("Job %s not found", job_id)
1161
      return None
1162

    
1163
    def _CheckForChanges():
1164
      logging.debug("Waiting for changes in job %s", job_id)
1165

    
1166
      status = job.CalcStatus()
1167
      job_info = self._GetJobInfoUnlocked(job, fields)
1168
      log_entries = job.GetLogEntries(prev_log_serial)
1169

    
1170
      # Serializing and deserializing data can cause type changes (e.g. from
1171
      # tuple to list) or precision loss. We're doing it here so that we get
1172
      # the same modifications as the data received from the client. Without
1173
      # this, the comparison afterwards might fail without the data being
1174
      # significantly different.
1175
      job_info = serializer.LoadJson(serializer.DumpJson(job_info))
1176
      log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
1177

    
1178
      # Don't even try to wait if the job is no longer running, there will be
1179
      # no changes.
1180
      if (status not in (constants.JOB_STATUS_QUEUED,
1181
                         constants.JOB_STATUS_RUNNING,
1182
                         constants.JOB_STATUS_WAITLOCK) or
1183
          prev_job_info != job_info or
1184
          (log_entries and prev_log_serial != log_entries[0][0])):
1185
        logging.debug("Job %s changed", job_id)
1186
        return (job_info, log_entries)
1187

    
1188
      raise utils.RetryAgain()
1189

    
1190
    try:
1191
      # Setting wait function to release the queue lock while waiting
1192
      return utils.Retry(_CheckForChanges, utils.RETRY_REMAINING_TIME, timeout,
1193
                         wait_fn=job.change.wait)
1194
    except utils.RetryTimeout:
1195
      return constants.JOB_NOTCHANGED
1196

    
1197
  @utils.LockedMethod
1198
  @_RequireOpenQueue
1199
  def CancelJob(self, job_id):
1200
    """Cancels a job.
1201

1202
    This will only succeed if the job has not started yet.
1203

1204
    @type job_id: string
1205
    @param job_id: job ID of job to be cancelled.
1206

1207
    """
1208
    logging.info("Cancelling job %s", job_id)
1209

    
1210
    job = self._LoadJobUnlocked(job_id)
1211
    if not job:
1212
      logging.debug("Job %s not found", job_id)
1213
      return (False, "Job %s not found" % job_id)
1214

    
1215
    job_status = job.CalcStatus()
1216

    
1217
    if job_status not in (constants.JOB_STATUS_QUEUED,
1218
                          constants.JOB_STATUS_WAITLOCK):
1219
      logging.debug("Job %s is no longer waiting in the queue", job.id)
1220
      return (False, "Job %s is no longer waiting in the queue" % job.id)
1221

    
1222
    if job_status == constants.JOB_STATUS_QUEUED:
1223
      self.CancelJobUnlocked(job)
1224
      return (True, "Job %s canceled" % job.id)
1225

    
1226
    elif job_status == constants.JOB_STATUS_WAITLOCK:
1227
      # The worker will notice the new status and cancel the job
1228
      try:
1229
        job.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
1230
      finally:
1231
        self.UpdateJobUnlocked(job)
1232
      return (True, "Job %s will be canceled" % job.id)
1233

    
1234
  @_RequireOpenQueue
1235
  def CancelJobUnlocked(self, job):
1236
    """Marks a job as canceled.
1237

1238
    """
1239
    try:
1240
      job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
1241
                            "Job canceled by request")
1242
    finally:
1243
      self.UpdateJobUnlocked(job)
1244

    
1245
  @_RequireOpenQueue
1246
  def _ArchiveJobsUnlocked(self, jobs):
1247
    """Archives jobs.
1248

1249
    @type jobs: list of L{_QueuedJob}
1250
    @param jobs: Job objects
1251
    @rtype: int
1252
    @return: Number of archived jobs
1253

1254
    """
1255
    archive_jobs = []
1256
    rename_files = []
1257
    for job in jobs:
1258
      if job.CalcStatus() not in (constants.JOB_STATUS_CANCELED,
1259
                                  constants.JOB_STATUS_SUCCESS,
1260
                                  constants.JOB_STATUS_ERROR):
1261
        logging.debug("Job %s is not yet done", job.id)
1262
        continue
1263

    
1264
      archive_jobs.append(job)
1265

    
1266
      old = self._GetJobPath(job.id)
1267
      new = self._GetArchivedJobPath(job.id)
1268
      rename_files.append((old, new))
1269

    
1270
    # TODO: What if 1..n files fail to rename?
1271
    self._RenameFilesUnlocked(rename_files)
1272

    
1273
    logging.debug("Successfully archived job(s) %s",
1274
                  utils.CommaJoin(job.id for job in archive_jobs))
1275

    
1276
    return len(archive_jobs)
1277

    
1278
  @utils.LockedMethod
1279
  @_RequireOpenQueue
1280
  def ArchiveJob(self, job_id):
1281
    """Archives a job.
1282

1283
    This is just a wrapper over L{_ArchiveJobsUnlocked}.
1284

1285
    @type job_id: string
1286
    @param job_id: Job ID of job to be archived.
1287
    @rtype: bool
1288
    @return: Whether job was archived
1289

1290
    """
1291
    logging.info("Archiving job %s", job_id)
1292

    
1293
    job = self._LoadJobUnlocked(job_id)
1294
    if not job:
1295
      logging.debug("Job %s not found", job_id)
1296
      return False
1297

    
1298
    return self._ArchiveJobsUnlocked([job]) == 1
1299

    
1300
  @utils.LockedMethod
1301
  @_RequireOpenQueue
1302
  def AutoArchiveJobs(self, age, timeout):
1303
    """Archives all jobs based on age.
1304

1305
    The method will archive all jobs which are older than the age
1306
    parameter. For jobs that don't have an end timestamp, the start
1307
    timestamp will be considered. The special '-1' age will cause
1308
    archival of all jobs (that are not running or queued).
1309

1310
    @type age: int
1311
    @param age: the minimum age in seconds
1312

1313
    """
1314
    logging.info("Archiving jobs with age more than %s seconds", age)
1315

    
1316
    now = time.time()
1317
    end_time = now + timeout
1318
    archived_count = 0
1319
    last_touched = 0
1320

    
1321
    all_job_ids = self._GetJobIDsUnlocked(archived=False)
1322
    pending = []
1323
    for idx, job_id in enumerate(all_job_ids):
1324
      last_touched = idx + 1
1325

    
1326
      # Not optimal because jobs could be pending
1327
      # TODO: Measure average duration for job archival and take number of
1328
      # pending jobs into account.
1329
      if time.time() > end_time:
1330
        break
1331

    
1332
      # Returns None if the job failed to load
1333
      job = self._LoadJobUnlocked(job_id)
1334
      if job:
1335
        if job.end_timestamp is None:
1336
          if job.start_timestamp is None:
1337
            job_age = job.received_timestamp
1338
          else:
1339
            job_age = job.start_timestamp
1340
        else:
1341
          job_age = job.end_timestamp
1342

    
1343
        if age == -1 or now - job_age[0] > age:
1344
          pending.append(job)
1345

    
1346
          # Archive 10 jobs at a time
1347
          if len(pending) >= 10:
1348
            archived_count += self._ArchiveJobsUnlocked(pending)
1349
            pending = []
1350

    
1351
    if pending:
1352
      archived_count += self._ArchiveJobsUnlocked(pending)
1353

    
1354
    return (archived_count, len(all_job_ids) - last_touched)
1355

    
1356
  @staticmethod
1357
  def _GetJobInfoUnlocked(job, fields):
1358
    """Returns information about a job.
1359

1360
    @type job: L{_QueuedJob}
1361
    @param job: the job which we query
1362
    @type fields: list
1363
    @param fields: names of fields to return
1364
    @rtype: list
1365
    @return: list with one element for each field
1366
    @raise errors.OpExecError: when an invalid field
1367
        has been passed
1368

1369
    """
1370
    row = []
1371
    for fname in fields:
1372
      if fname == "id":
1373
        row.append(job.id)
1374
      elif fname == "status":
1375
        row.append(job.CalcStatus())
1376
      elif fname == "ops":
1377
        row.append([op.input.__getstate__() for op in job.ops])
1378
      elif fname == "opresult":
1379
        row.append([op.result for op in job.ops])
1380
      elif fname == "opstatus":
1381
        row.append([op.status for op in job.ops])
1382
      elif fname == "oplog":
1383
        row.append([op.log for op in job.ops])
1384
      elif fname == "opstart":
1385
        row.append([op.start_timestamp for op in job.ops])
1386
      elif fname == "opend":
1387
        row.append([op.end_timestamp for op in job.ops])
1388
      elif fname == "received_ts":
1389
        row.append(job.received_timestamp)
1390
      elif fname == "start_ts":
1391
        row.append(job.start_timestamp)
1392
      elif fname == "end_ts":
1393
        row.append(job.end_timestamp)
1394
      elif fname == "lock_status":
1395
        row.append(job.lock_status)
1396
      elif fname == "summary":
1397
        row.append([op.input.Summary() for op in job.ops])
1398
      else:
1399
        raise errors.OpExecError("Invalid job query field '%s'" % fname)
1400
    return row
1401

    
1402
  @utils.LockedMethod
1403
  @_RequireOpenQueue
1404
  def QueryJobs(self, job_ids, fields):
1405
    """Returns a list of jobs in queue.
1406

1407
    This is a wrapper of L{_GetJobsUnlocked}, which actually does the
1408
    processing for each job.
1409

1410
    @type job_ids: list
1411
    @param job_ids: sequence of job identifiers or None for all
1412
    @type fields: list
1413
    @param fields: names of fields to return
1414
    @rtype: list
1415
    @return: list one element per job, each element being list with
1416
        the requested fields
1417

1418
    """
1419
    jobs = []
1420

    
1421
    for job in self._GetJobsUnlocked(job_ids):
1422
      if job is None:
1423
        jobs.append(None)
1424
      else:
1425
        jobs.append(self._GetJobInfoUnlocked(job, fields))
1426

    
1427
    return jobs
1428

    
1429
  @utils.LockedMethod
1430
  @_RequireOpenQueue
1431
  def Shutdown(self):
1432
    """Stops the job queue.
1433

1434
    This shutdowns all the worker threads an closes the queue.
1435

1436
    """
1437
    self._wpool.TerminateWorkers()
1438

    
1439
    self._queue_lock.Close()
1440
    self._queue_lock = None