Statistics
| Branch: | Tag: | Revision:

root / lib / jqueue.py @ 85a1c57d

History | View | Annotate | Download (41.4 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 exec_timestamp: timestamp for the actual LU Exec() function invocation
81
  @ivar stop_timestamp: timestamp for the end of the execution
82

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

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

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

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

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

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

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

    
123
  def Serialize(self):
124
    """Serializes this _QueuedOpCode.
125

126
    @rtype: dict
127
    @return: the dictionary holding the serialized state
128

129
    """
130
    return {
131
      "input": self.input.__getstate__(),
132
      "status": self.status,
133
      "result": self.result,
134
      "log": self.log,
135
      "start_timestamp": self.start_timestamp,
136
      "exec_timestamp": self.exec_timestamp,
137
      "end_timestamp": self.end_timestamp,
138
      }
139

    
140

    
141
class _QueuedJob(object):
142
  """In-memory job representation.
143

144
  This is what we use to track the user-submitted jobs. Locking must
145
  be taken care of by users of this class.
146

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

160
  """
161
  # pylint: disable-msg=W0212
162
  __slots__ = ["queue", "id", "ops", "log_serial",
163
               "received_timestamp", "start_timestamp", "end_timestamp",
164
               "lock_status", "change",
165
               "__weakref__"]
166

    
167
  def __init__(self, queue, job_id, ops):
168
    """Constructor for the _QueuedJob.
169

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

178
    """
179
    if not ops:
180
      raise errors.GenericError("A job needs at least one opcode")
181

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

    
190
    # In-memory attributes
191
    self.lock_status = None
192

    
193
    # Condition to wait for changes
194
    self.change = threading.Condition(self.queue._lock)
195

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

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

    
203
  @classmethod
204
  def Restore(cls, queue, state):
205
    """Restore a _QueuedJob from serialized state:
206

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

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

    
222
    # In-memory attributes
223
    obj.lock_status = None
224

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

    
233
    # Condition to wait for changes
234
    obj.change = threading.Condition(obj.queue._lock)
235

    
236
    return obj
237

    
238
  def Serialize(self):
239
    """Serialize the _JobQueue instance.
240

241
    @rtype: dict
242
    @return: the serialized state
243

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

    
253
  def CalcStatus(self):
254
    """Compute the status of this job.
255

256
    This function iterates over all the _QueuedOpCodes in the job and
257
    based on their status, computes the job status.
258

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

267
        will determine the job status
268

269
      - otherwise, it means either all opcodes are queued, or success,
270
        and the job status will be the same
271

272
    @return: the job status
273

274
    """
275
    status = constants.JOB_STATUS_QUEUED
276

    
277
    all_success = True
278
    for op in self.ops:
279
      if op.status == constants.OP_STATUS_SUCCESS:
280
        continue
281

    
282
      all_success = False
283

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

    
301
    if all_success:
302
      status = constants.JOB_STATUS_SUCCESS
303

    
304
    return status
305

    
306
  def GetLogEntries(self, newer_than):
307
    """Selectively returns the log entries.
308

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

316
    """
317
    if newer_than is None:
318
      serial = -1
319
    else:
320
      serial = newer_than
321

    
322
    entries = []
323
    for op in self.ops:
324
      entries.extend(filter(lambda entry: entry[0] > serial, op.log))
325

    
326
    return entries
327

    
328
  def MarkUnfinishedOps(self, status, result):
329
    """Mark unfinished opcodes with a given status and result.
330

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

335
    @param status: a given opcode status
336
    @param result: the opcode result
337

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

    
348

    
349
class _OpExecCallbacks(mcpu.OpExecCbBase):
350
  def __init__(self, queue, job, op):
351
    """Initializes this class.
352

353
    @type queue: L{JobQueue}
354
    @param queue: Job queue
355
    @type job: L{_QueuedJob}
356
    @param job: Job object
357
    @type op: L{_QueuedOpCode}
358
    @param op: OpCode
359

360
    """
361
    assert queue, "Queue is missing"
362
    assert job, "Job is missing"
363
    assert op, "Opcode is missing"
364

    
365
    self._queue = queue
366
    self._job = job
367
    self._op = op
368

    
369
  def NotifyStart(self):
370
    """Mark the opcode as running, not lock-waiting.
371

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

377
    """
378
    self._queue.acquire()
379
    try:
380
      assert self._op.status in (constants.OP_STATUS_WAITLOCK,
381
                                 constants.OP_STATUS_CANCELING)
382

    
383
      # All locks are acquired by now
384
      self._job.lock_status = None
385

    
386
      # Cancel here if we were asked to
387
      if self._op.status == constants.OP_STATUS_CANCELING:
388
        raise CancelJob()
389

    
390
      self._op.status = constants.OP_STATUS_RUNNING
391
      self._op.exec_timestamp = TimeStampNow()
392
    finally:
393
      self._queue.release()
394

    
395
  def Feedback(self, *args):
396
    """Append a log entry.
397

398
    """
399
    assert len(args) < 3
400

    
401
    if len(args) == 1:
402
      log_type = constants.ELOG_MESSAGE
403
      log_msg = args[0]
404
    else:
405
      (log_type, log_msg) = args
406

    
407
    # The time is split to make serialization easier and not lose
408
    # precision.
409
    timestamp = utils.SplitTime(time.time())
410

    
411
    self._queue.acquire()
412
    try:
413
      self._job.log_serial += 1
414
      self._op.log.append((self._job.log_serial, timestamp, log_type, log_msg))
415

    
416
      self._job.change.notifyAll()
417
    finally:
418
      self._queue.release()
419

    
420
  def ReportLocks(self, msg):
421
    """Write locking information to the job.
422

423
    Called whenever the LU processor is waiting for a lock or has acquired one.
424

425
    """
426
    # Not getting the queue lock because this is a single assignment
427
    self._job.lock_status = msg
428

    
429

    
430
class _JobQueueWorker(workerpool.BaseWorker):
431
  """The actual job workers.
432

433
  """
434
  def RunTask(self, job): # pylint: disable-msg=W0221
435
    """Job executor.
436

437
    This functions processes a job. It is closely tied to the _QueuedJob and
438
    _QueuedOpCode classes.
439

440
    @type job: L{_QueuedJob}
441
    @param job: the job to be processed
442

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

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

    
477
              input_opcode = op.input
478
            finally:
479
              queue.release()
480

    
481
            # Make sure not to hold queue lock while calling ExecOpCode
482
            result = proc.ExecOpCode(input_opcode,
483
                                     _OpExecCallbacks(queue, job, op))
484

    
485
            queue.acquire()
486
            try:
487
              op.status = constants.OP_STATUS_SUCCESS
488
              op.result = result
489
              op.end_timestamp = TimeStampNow()
490
              queue.UpdateJobUnlocked(job)
491
            finally:
492
              queue.release()
493

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

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

    
540
      logging.info("Finished job %s, status = %s", job_id, status)
541

    
542

    
543
class _JobQueueWorkerPool(workerpool.WorkerPool):
544
  """Simple class implementing a job-processing workerpool.
545

546
  """
547
  def __init__(self, queue):
548
    super(_JobQueueWorkerPool, self).__init__("JobQueue",
549
                                              JOBQUEUE_THREADS,
550
                                              _JobQueueWorker)
551
    self.queue = queue
552

    
553

    
554
def _RequireOpenQueue(fn):
555
  """Decorator for "public" functions.
556

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

563
  @warning: Use this decorator only after utils.LockedMethod!
564

565
  Example::
566
    @utils.LockedMethod
567
    @_RequireOpenQueue
568
    def Example(self):
569
      pass
570

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

    
578

    
579
class JobQueue(object):
580
  """Queue used to manage the jobs.
581

582
  @cvar _RE_JOB_FILE: regex matching the valid job file names
583

584
  """
585
  _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
586

    
587
  def __init__(self, context):
588
    """Constructor for JobQueue.
589

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

595
    @type context: GanetiContext
596
    @param context: the context object for access to the configuration
597
        data and other ganeti objects
598

599
    """
600
    self.context = context
601
    self._memcache = weakref.WeakValueDictionary()
602
    self._my_hostname = utils.HostInfo().name
603

    
604
    # Locking
605
    self._lock = threading.Lock()
606
    self.acquire = self._lock.acquire
607
    self.release = self._lock.release
608

    
609
    # Initialize
610
    self._queue_lock = jstore.InitAndVerifyQueue(must_lock=True)
611

    
612
    # Read serial file
613
    self._last_serial = jstore.ReadSerial()
614
    assert self._last_serial is not None, ("Serial file was modified between"
615
                                           " check in jstore and here")
616

    
617
    # Get initial list of nodes
618
    self._nodes = dict((n.name, n.primary_ip)
619
                       for n in self.context.cfg.GetAllNodesInfo().values()
620
                       if n.master_candidate)
621

    
622
    # Remove master node
623
    try:
624
      del self._nodes[self._my_hostname]
625
    except KeyError:
626
      pass
627

    
628
    # TODO: Check consistency across nodes
629

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

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

    
650
          job = self._LoadJobUnlocked(job_id)
651

    
652
          # a failure in loading the job can cause 'None' to be returned
653
          if job is None:
654
            continue
655

    
656
          status = job.CalcStatus()
657

    
658
          if status in (constants.JOB_STATUS_QUEUED, ):
659
            self._wpool.AddTask(job)
660

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

    
671
        logging.info("Job queue inspection finished")
672
      finally:
673
        self.release()
674
    except:
675
      self._wpool.TerminateWorkers()
676
      raise
677

    
678
  @utils.LockedMethod
679
  @_RequireOpenQueue
680
  def AddNode(self, node):
681
    """Register a new node with the queue.
682

683
    @type node: L{objects.Node}
684
    @param node: the node object to be added
685

686
    """
687
    node_name = node.name
688
    assert node_name != self._my_hostname
689

    
690
    # Clean queue directory on added node
691
    result = rpc.RpcRunner.call_jobqueue_purge(node_name)
692
    msg = result.fail_msg
693
    if msg:
694
      logging.warning("Cannot cleanup queue directory on node %s: %s",
695
                      node_name, msg)
696

    
697
    if not node.master_candidate:
698
      # remove if existing, ignoring errors
699
      self._nodes.pop(node_name, None)
700
      # and skip the replication of the job ids
701
      return
702

    
703
    # Upload the whole queue excluding archived jobs
704
    files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
705

    
706
    # Upload current serial file
707
    files.append(constants.JOB_QUEUE_SERIAL_FILE)
708

    
709
    for file_name in files:
710
      # Read file content
711
      content = utils.ReadFile(file_name)
712

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

    
721
    self._nodes[node_name] = node.primary_ip
722

    
723
  @utils.LockedMethod
724
  @_RequireOpenQueue
725
  def RemoveNode(self, node_name):
726
    """Callback called when removing nodes from the cluster.
727

728
    @type node_name: str
729
    @param node_name: the name of the node to remove
730

731
    """
732
    try:
733
      # The queue is removed by the "leave node" RPC call.
734
      del self._nodes[node_name]
735
    except KeyError:
736
      pass
737

    
738
  @staticmethod
739
  def _CheckRpcResult(result, nodes, failmsg):
740
    """Verifies the status of an RPC call.
741

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

746
    @param result: the data as returned from the rpc call
747
    @type nodes: list
748
    @param nodes: the list of nodes we made the call to
749
    @type failmsg: str
750
    @param failmsg: the identifier to be used for logging
751

752
    """
753
    failed = []
754
    success = []
755

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

    
765
    # +1 for the master node
766
    if (len(success) + 1) < len(failed):
767
      # TODO: Handle failing nodes
768
      logging.error("More than half of the nodes failed")
769

    
770
  def _GetNodeIp(self):
771
    """Helper for returning the node name/ip list.
772

773
    @rtype: (list, list)
774
    @return: a tuple of two lists, the first one with the node
775
        names and the second one with the node addresses
776

777
    """
778
    name_list = self._nodes.keys()
779
    addr_list = [self._nodes[name] for name in name_list]
780
    return name_list, addr_list
781

    
782
  def _WriteAndReplicateFileUnlocked(self, file_name, data):
783
    """Writes a file locally and then replicates it to all nodes.
784

785
    This function will replace the contents of a file on the local
786
    node and then replicate it to all the other nodes we have.
787

788
    @type file_name: str
789
    @param file_name: the path of the file to be replicated
790
    @type data: str
791
    @param data: the new contents of the file
792

793
    """
794
    utils.WriteFile(file_name, data=data)
795

    
796
    names, addrs = self._GetNodeIp()
797
    result = rpc.RpcRunner.call_jobqueue_update(names, addrs, file_name, data)
798
    self._CheckRpcResult(result, self._nodes,
799
                         "Updating %s" % file_name)
800

    
801
  def _RenameFilesUnlocked(self, rename):
802
    """Renames a file locally and then replicate the change.
803

804
    This function will rename a file in the local queue directory
805
    and then replicate this rename to all the other nodes we have.
806

807
    @type rename: list of (old, new)
808
    @param rename: List containing tuples mapping old to new names
809

810
    """
811
    # Rename them locally
812
    for old, new in rename:
813
      utils.RenameFile(old, new, mkdir=True)
814

    
815
    # ... and on all nodes
816
    names, addrs = self._GetNodeIp()
817
    result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
818
    self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
819

    
820
  @staticmethod
821
  def _FormatJobID(job_id):
822
    """Convert a job ID to string format.
823

824
    Currently this just does C{str(job_id)} after performing some
825
    checks, but if we want to change the job id format this will
826
    abstract this change.
827

828
    @type job_id: int or long
829
    @param job_id: the numeric job id
830
    @rtype: str
831
    @return: the formatted job id
832

833
    """
834
    if not isinstance(job_id, (int, long)):
835
      raise errors.ProgrammerError("Job ID '%s' not numeric" % job_id)
836
    if job_id < 0:
837
      raise errors.ProgrammerError("Job ID %s is negative" % job_id)
838

    
839
    return str(job_id)
840

    
841
  @classmethod
842
  def _GetArchiveDirectory(cls, job_id):
843
    """Returns the archive directory for a job.
844

845
    @type job_id: str
846
    @param job_id: Job identifier
847
    @rtype: str
848
    @return: Directory name
849

850
    """
851
    return str(int(job_id) / JOBS_PER_ARCHIVE_DIRECTORY)
852

    
853
  def _NewSerialsUnlocked(self, count):
854
    """Generates a new job identifier.
855

856
    Job identifiers are unique during the lifetime of a cluster.
857

858
    @type count: integer
859
    @param count: how many serials to return
860
    @rtype: str
861
    @return: a string representing the job identifier.
862

863
    """
864
    assert count > 0
865
    # New number
866
    serial = self._last_serial + count
867

    
868
    # Write to file
869
    self._WriteAndReplicateFileUnlocked(constants.JOB_QUEUE_SERIAL_FILE,
870
                                        "%s\n" % serial)
871

    
872
    result = [self._FormatJobID(v)
873
              for v in range(self._last_serial, serial + 1)]
874
    # Keep it only if we were able to write the file
875
    self._last_serial = serial
876

    
877
    return result
878

    
879
  @staticmethod
880
  def _GetJobPath(job_id):
881
    """Returns the job file for a given job id.
882

883
    @type job_id: str
884
    @param job_id: the job identifier
885
    @rtype: str
886
    @return: the path to the job file
887

888
    """
889
    return utils.PathJoin(constants.QUEUE_DIR, "job-%s" % job_id)
890

    
891
  @classmethod
892
  def _GetArchivedJobPath(cls, job_id):
893
    """Returns the archived job file for a give job id.
894

895
    @type job_id: str
896
    @param job_id: the job identifier
897
    @rtype: str
898
    @return: the path to the archived job file
899

900
    """
901
    return utils.PathJoin(constants.JOB_QUEUE_ARCHIVE_DIR,
902
                          cls._GetArchiveDirectory(job_id), "job-%s" % job_id)
903

    
904
  def _GetJobIDsUnlocked(self, sort=True):
905
    """Return all known job IDs.
906

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

911
    @type sort: boolean
912
    @param sort: perform sorting on the returned job ids
913
    @rtype: list
914
    @return: the list of job IDs
915

916
    """
917
    jlist = []
918
    for filename in utils.ListVisibleFiles(constants.QUEUE_DIR, sort=False):
919
      m = self._RE_JOB_FILE.match(filename)
920
      if m:
921
        jlist.append(m.group(1))
922
    if sort:
923
      jlist = utils.NiceSort(jlist)
924
    return jlist
925

    
926
  def _LoadJobUnlocked(self, job_id):
927
    """Loads a job from the disk or memory.
928

929
    Given a job id, this will return the cached job object if
930
    existing, or try to load the job from the disk. If loading from
931
    disk, it will also add the job to the cache.
932

933
    @param job_id: the job id
934
    @rtype: L{_QueuedJob} or None
935
    @return: either None or the job object
936

937
    """
938
    job = self._memcache.get(job_id, None)
939
    if job:
940
      logging.debug("Found job %s in memcache", job_id)
941
      return job
942

    
943
    filepath = self._GetJobPath(job_id)
944
    logging.debug("Loading job from %s", filepath)
945
    try:
946
      raw_data = utils.ReadFile(filepath)
947
    except IOError, err:
948
      if err.errno in (errno.ENOENT, ):
949
        return None
950
      raise
951

    
952
    data = serializer.LoadJson(raw_data)
953

    
954
    try:
955
      job = _QueuedJob.Restore(self, data)
956
    except Exception, err: # pylint: disable-msg=W0703
957
      new_path = self._GetArchivedJobPath(job_id)
958
      if filepath == new_path:
959
        # job already archived (future case)
960
        logging.exception("Can't parse job %s", job_id)
961
      else:
962
        # non-archived case
963
        logging.exception("Can't parse job %s, will archive.", job_id)
964
        self._RenameFilesUnlocked([(filepath, new_path)])
965
      return None
966

    
967
    self._memcache[job_id] = job
968
    logging.debug("Added job %s to the cache", job_id)
969
    return job
970

    
971
  def _GetJobsUnlocked(self, job_ids):
972
    """Return a list of jobs based on their IDs.
973

974
    @type job_ids: list
975
    @param job_ids: either an empty list (meaning all jobs),
976
        or a list of job IDs
977
    @rtype: list
978
    @return: the list of job objects
979

980
    """
981
    if not job_ids:
982
      job_ids = self._GetJobIDsUnlocked()
983

    
984
    return [self._LoadJobUnlocked(job_id) for job_id in job_ids]
985

    
986
  @staticmethod
987
  def _IsQueueMarkedDrain():
988
    """Check if the queue is marked from drain.
989

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

993
    @rtype: boolean
994
    @return: True of the job queue is marked for draining
995

996
    """
997
    return os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
998

    
999
  @staticmethod
1000
  def SetDrainFlag(drain_flag):
1001
    """Sets the drain flag for the queue.
1002

1003
    @type drain_flag: boolean
1004
    @param drain_flag: Whether to set or unset the drain flag
1005

1006
    """
1007
    if drain_flag:
1008
      utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
1009
    else:
1010
      utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
1011
    return True
1012

    
1013
  @_RequireOpenQueue
1014
  def _SubmitJobUnlocked(self, job_id, ops):
1015
    """Create and store a new job.
1016

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

1020
    @type job_id: job ID
1021
    @param job_id: the job ID for the new job
1022
    @type ops: list
1023
    @param ops: The list of OpCodes that will become the new job.
1024
    @rtype: job ID
1025
    @return: the job ID of the newly created job
1026
    @raise errors.JobQueueDrainError: if the job is marked for draining
1027

1028
    """
1029
    if self._IsQueueMarkedDrain():
1030
      raise errors.JobQueueDrainError("Job queue is drained, refusing job")
1031

    
1032
    # Check job queue size
1033
    size = len(self._GetJobIDsUnlocked(sort=False))
1034
    if size >= constants.JOB_QUEUE_SIZE_SOFT_LIMIT:
1035
      # TODO: Autoarchive jobs. Make sure it's not done on every job
1036
      # submission, though.
1037
      #size = ...
1038
      pass
1039

    
1040
    if size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
1041
      raise errors.JobQueueFull()
1042

    
1043
    job = _QueuedJob(self, job_id, ops)
1044

    
1045
    # Write to disk
1046
    self.UpdateJobUnlocked(job)
1047

    
1048
    logging.debug("Adding new job %s to the cache", job_id)
1049
    self._memcache[job_id] = job
1050

    
1051
    # Add to worker pool
1052
    self._wpool.AddTask(job)
1053

    
1054
    return job.id
1055

    
1056
  @utils.LockedMethod
1057
  @_RequireOpenQueue
1058
  def SubmitJob(self, ops):
1059
    """Create and store a new job.
1060

1061
    @see: L{_SubmitJobUnlocked}
1062

1063
    """
1064
    job_id = self._NewSerialsUnlocked(1)[0]
1065
    return self._SubmitJobUnlocked(job_id, ops)
1066

    
1067
  @utils.LockedMethod
1068
  @_RequireOpenQueue
1069
  def SubmitManyJobs(self, jobs):
1070
    """Create and store multiple jobs.
1071

1072
    @see: L{_SubmitJobUnlocked}
1073

1074
    """
1075
    results = []
1076
    all_job_ids = self._NewSerialsUnlocked(len(jobs))
1077
    for job_id, ops in zip(all_job_ids, jobs):
1078
      try:
1079
        data = self._SubmitJobUnlocked(job_id, ops)
1080
        status = True
1081
      except errors.GenericError, err:
1082
        data = str(err)
1083
        status = False
1084
      results.append((status, data))
1085

    
1086
    return results
1087

    
1088
  @_RequireOpenQueue
1089
  def UpdateJobUnlocked(self, job):
1090
    """Update a job's on disk storage.
1091

1092
    After a job has been modified, this function needs to be called in
1093
    order to write the changes to disk and replicate them to the other
1094
    nodes.
1095

1096
    @type job: L{_QueuedJob}
1097
    @param job: the changed job
1098

1099
    """
1100
    filename = self._GetJobPath(job.id)
1101
    data = serializer.DumpJson(job.Serialize(), indent=False)
1102
    logging.debug("Writing job %s to %s", job.id, filename)
1103
    self._WriteAndReplicateFileUnlocked(filename, data)
1104

    
1105
    # Notify waiters about potential changes
1106
    job.change.notifyAll()
1107

    
1108
  @utils.LockedMethod
1109
  @_RequireOpenQueue
1110
  def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial,
1111
                        timeout):
1112
    """Waits for changes in a job.
1113

1114
    @type job_id: string
1115
    @param job_id: Job identifier
1116
    @type fields: list of strings
1117
    @param fields: Which fields to check for changes
1118
    @type prev_job_info: list or None
1119
    @param prev_job_info: Last job information returned
1120
    @type prev_log_serial: int
1121
    @param prev_log_serial: Last job message serial number
1122
    @type timeout: float
1123
    @param timeout: maximum time to wait
1124
    @rtype: tuple (job info, log entries)
1125
    @return: a tuple of the job information as required via
1126
        the fields parameter, and the log entries as a list
1127

1128
        if the job has not changed and the timeout has expired,
1129
        we instead return a special value,
1130
        L{constants.JOB_NOTCHANGED}, which should be interpreted
1131
        as such by the clients
1132

1133
    """
1134
    job = self._LoadJobUnlocked(job_id)
1135
    if not job:
1136
      logging.debug("Job %s not found", job_id)
1137
      return None
1138

    
1139
    def _CheckForChanges():
1140
      logging.debug("Waiting for changes in job %s", job_id)
1141

    
1142
      status = job.CalcStatus()
1143
      job_info = self._GetJobInfoUnlocked(job, fields)
1144
      log_entries = job.GetLogEntries(prev_log_serial)
1145

    
1146
      # Serializing and deserializing data can cause type changes (e.g. from
1147
      # tuple to list) or precision loss. We're doing it here so that we get
1148
      # the same modifications as the data received from the client. Without
1149
      # this, the comparison afterwards might fail without the data being
1150
      # significantly different.
1151
      job_info = serializer.LoadJson(serializer.DumpJson(job_info))
1152
      log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
1153

    
1154
      # Don't even try to wait if the job is no longer running, there will be
1155
      # no changes.
1156
      if (status not in (constants.JOB_STATUS_QUEUED,
1157
                         constants.JOB_STATUS_RUNNING,
1158
                         constants.JOB_STATUS_WAITLOCK) or
1159
          prev_job_info != job_info or
1160
          (log_entries and prev_log_serial != log_entries[0][0])):
1161
        logging.debug("Job %s changed", job_id)
1162
        return (job_info, log_entries)
1163

    
1164
      raise utils.RetryAgain()
1165

    
1166
    try:
1167
      # Setting wait function to release the queue lock while waiting
1168
      return utils.Retry(_CheckForChanges, utils.RETRY_REMAINING_TIME, timeout,
1169
                         wait_fn=job.change.wait)
1170
    except utils.RetryTimeout:
1171
      return constants.JOB_NOTCHANGED
1172

    
1173
  @utils.LockedMethod
1174
  @_RequireOpenQueue
1175
  def CancelJob(self, job_id):
1176
    """Cancels a job.
1177

1178
    This will only succeed if the job has not started yet.
1179

1180
    @type job_id: string
1181
    @param job_id: job ID of job to be cancelled.
1182

1183
    """
1184
    logging.info("Cancelling job %s", job_id)
1185

    
1186
    job = self._LoadJobUnlocked(job_id)
1187
    if not job:
1188
      logging.debug("Job %s not found", job_id)
1189
      return (False, "Job %s not found" % job_id)
1190

    
1191
    job_status = job.CalcStatus()
1192

    
1193
    if job_status not in (constants.JOB_STATUS_QUEUED,
1194
                          constants.JOB_STATUS_WAITLOCK):
1195
      logging.debug("Job %s is no longer waiting in the queue", job.id)
1196
      return (False, "Job %s is no longer waiting in the queue" % job.id)
1197

    
1198
    if job_status == constants.JOB_STATUS_QUEUED:
1199
      self.CancelJobUnlocked(job)
1200
      return (True, "Job %s canceled" % job.id)
1201

    
1202
    elif job_status == constants.JOB_STATUS_WAITLOCK:
1203
      # The worker will notice the new status and cancel the job
1204
      try:
1205
        job.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
1206
      finally:
1207
        self.UpdateJobUnlocked(job)
1208
      return (True, "Job %s will be canceled" % job.id)
1209

    
1210
  @_RequireOpenQueue
1211
  def CancelJobUnlocked(self, job):
1212
    """Marks a job as canceled.
1213

1214
    """
1215
    try:
1216
      job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
1217
                            "Job canceled by request")
1218
    finally:
1219
      self.UpdateJobUnlocked(job)
1220

    
1221
  @_RequireOpenQueue
1222
  def _ArchiveJobsUnlocked(self, jobs):
1223
    """Archives jobs.
1224

1225
    @type jobs: list of L{_QueuedJob}
1226
    @param jobs: Job objects
1227
    @rtype: int
1228
    @return: Number of archived jobs
1229

1230
    """
1231
    archive_jobs = []
1232
    rename_files = []
1233
    for job in jobs:
1234
      if job.CalcStatus() not in (constants.JOB_STATUS_CANCELED,
1235
                                  constants.JOB_STATUS_SUCCESS,
1236
                                  constants.JOB_STATUS_ERROR):
1237
        logging.debug("Job %s is not yet done", job.id)
1238
        continue
1239

    
1240
      archive_jobs.append(job)
1241

    
1242
      old = self._GetJobPath(job.id)
1243
      new = self._GetArchivedJobPath(job.id)
1244
      rename_files.append((old, new))
1245

    
1246
    # TODO: What if 1..n files fail to rename?
1247
    self._RenameFilesUnlocked(rename_files)
1248

    
1249
    logging.debug("Successfully archived job(s) %s",
1250
                  utils.CommaJoin(job.id for job in archive_jobs))
1251

    
1252
    return len(archive_jobs)
1253

    
1254
  @utils.LockedMethod
1255
  @_RequireOpenQueue
1256
  def ArchiveJob(self, job_id):
1257
    """Archives a job.
1258

1259
    This is just a wrapper over L{_ArchiveJobsUnlocked}.
1260

1261
    @type job_id: string
1262
    @param job_id: Job ID of job to be archived.
1263
    @rtype: bool
1264
    @return: Whether job was archived
1265

1266
    """
1267
    logging.info("Archiving job %s", job_id)
1268

    
1269
    job = self._LoadJobUnlocked(job_id)
1270
    if not job:
1271
      logging.debug("Job %s not found", job_id)
1272
      return False
1273

    
1274
    return self._ArchiveJobsUnlocked([job]) == 1
1275

    
1276
  @utils.LockedMethod
1277
  @_RequireOpenQueue
1278
  def AutoArchiveJobs(self, age, timeout):
1279
    """Archives all jobs based on age.
1280

1281
    The method will archive all jobs which are older than the age
1282
    parameter. For jobs that don't have an end timestamp, the start
1283
    timestamp will be considered. The special '-1' age will cause
1284
    archival of all jobs (that are not running or queued).
1285

1286
    @type age: int
1287
    @param age: the minimum age in seconds
1288

1289
    """
1290
    logging.info("Archiving jobs with age more than %s seconds", age)
1291

    
1292
    now = time.time()
1293
    end_time = now + timeout
1294
    archived_count = 0
1295
    last_touched = 0
1296

    
1297
    all_job_ids = self._GetJobIDsUnlocked()
1298
    pending = []
1299
    for idx, job_id in enumerate(all_job_ids):
1300
      last_touched = idx + 1
1301

    
1302
      # Not optimal because jobs could be pending
1303
      # TODO: Measure average duration for job archival and take number of
1304
      # pending jobs into account.
1305
      if time.time() > end_time:
1306
        break
1307

    
1308
      # Returns None if the job failed to load
1309
      job = self._LoadJobUnlocked(job_id)
1310
      if job:
1311
        if job.end_timestamp is None:
1312
          if job.start_timestamp is None:
1313
            job_age = job.received_timestamp
1314
          else:
1315
            job_age = job.start_timestamp
1316
        else:
1317
          job_age = job.end_timestamp
1318

    
1319
        if age == -1 or now - job_age[0] > age:
1320
          pending.append(job)
1321

    
1322
          # Archive 10 jobs at a time
1323
          if len(pending) >= 10:
1324
            archived_count += self._ArchiveJobsUnlocked(pending)
1325
            pending = []
1326

    
1327
    if pending:
1328
      archived_count += self._ArchiveJobsUnlocked(pending)
1329

    
1330
    return (archived_count, len(all_job_ids) - last_touched)
1331

    
1332
  @staticmethod
1333
  def _GetJobInfoUnlocked(job, fields):
1334
    """Returns information about a job.
1335

1336
    @type job: L{_QueuedJob}
1337
    @param job: the job which we query
1338
    @type fields: list
1339
    @param fields: names of fields to return
1340
    @rtype: list
1341
    @return: list with one element for each field
1342
    @raise errors.OpExecError: when an invalid field
1343
        has been passed
1344

1345
    """
1346
    row = []
1347
    for fname in fields:
1348
      if fname == "id":
1349
        row.append(job.id)
1350
      elif fname == "status":
1351
        row.append(job.CalcStatus())
1352
      elif fname == "ops":
1353
        row.append([op.input.__getstate__() for op in job.ops])
1354
      elif fname == "opresult":
1355
        row.append([op.result for op in job.ops])
1356
      elif fname == "opstatus":
1357
        row.append([op.status for op in job.ops])
1358
      elif fname == "oplog":
1359
        row.append([op.log for op in job.ops])
1360
      elif fname == "opstart":
1361
        row.append([op.start_timestamp for op in job.ops])
1362
      elif fname == "opexec":
1363
        row.append([op.exec_timestamp for op in job.ops])
1364
      elif fname == "opend":
1365
        row.append([op.end_timestamp for op in job.ops])
1366
      elif fname == "received_ts":
1367
        row.append(job.received_timestamp)
1368
      elif fname == "start_ts":
1369
        row.append(job.start_timestamp)
1370
      elif fname == "end_ts":
1371
        row.append(job.end_timestamp)
1372
      elif fname == "lock_status":
1373
        row.append(job.lock_status)
1374
      elif fname == "summary":
1375
        row.append([op.input.Summary() for op in job.ops])
1376
      else:
1377
        raise errors.OpExecError("Invalid job query field '%s'" % fname)
1378
    return row
1379

    
1380
  @utils.LockedMethod
1381
  @_RequireOpenQueue
1382
  def QueryJobs(self, job_ids, fields):
1383
    """Returns a list of jobs in queue.
1384

1385
    This is a wrapper of L{_GetJobsUnlocked}, which actually does the
1386
    processing for each job.
1387

1388
    @type job_ids: list
1389
    @param job_ids: sequence of job identifiers or None for all
1390
    @type fields: list
1391
    @param fields: names of fields to return
1392
    @rtype: list
1393
    @return: list one element per job, each element being list with
1394
        the requested fields
1395

1396
    """
1397
    jobs = []
1398

    
1399
    for job in self._GetJobsUnlocked(job_ids):
1400
      if job is None:
1401
        jobs.append(None)
1402
      else:
1403
        jobs.append(self._GetJobInfoUnlocked(job, fields))
1404

    
1405
    return jobs
1406

    
1407
  @utils.LockedMethod
1408
  @_RequireOpenQueue
1409
  def Shutdown(self):
1410
    """Stops the job queue.
1411

1412
    This shutdowns all the worker threads an closes the queue.
1413

1414
    """
1415
    self._wpool.TerminateWorkers()
1416

    
1417
    self._queue_lock.Close()
1418
    self._queue_lock = None