Statistics
| Branch: | Tag: | Revision:

root / lib / jqueue.py @ bcb66fca

History | View | Annotate | Download (40.5 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 run_op_index: int
149
  @ivar run_op_index: the currently executing opcode, or -1 if
150
      we didn't yet start executing
151
  @type log_serial: int
152
  @ivar log_serial: holds the index for the next log entry
153
  @ivar received_timestamp: the timestamp for when the job was received
154
  @ivar start_timestmap: the timestamp for start of execution
155
  @ivar end_timestamp: the timestamp for end of execution
156
  @ivar change: a Condition variable we use for waiting for job changes
157

158
  """
159
  __slots__ = ["queue", "id", "ops", "run_op_index", "log_serial",
160
               "received_timestamp", "start_timestamp", "end_timestamp",
161
               "change",
162
               "__weakref__"]
163

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

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

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

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

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

    
192
  @classmethod
193
  def Restore(cls, queue, state):
194
    """Restore a _QueuedJob from serialized state:
195

196
    @type queue: L{JobQueue}
197
    @param queue: to which queue the restored job belongs
198
    @type state: dict
199
    @param state: the serialized state
200
    @rtype: _JobQueue
201
    @return: the restored _JobQueue instance
202

203
    """
204
    obj = _QueuedJob.__new__(cls)
205
    obj.queue = queue
206
    obj.id = state["id"]
207
    obj.run_op_index = state["run_op_index"]
208
    obj.received_timestamp = state.get("received_timestamp", None)
209
    obj.start_timestamp = state.get("start_timestamp", None)
210
    obj.end_timestamp = state.get("end_timestamp", None)
211

    
212
    obj.ops = []
213
    obj.log_serial = 0
214
    for op_state in state["ops"]:
215
      op = _QueuedOpCode.Restore(op_state)
216
      for log_entry in op.log:
217
        obj.log_serial = max(obj.log_serial, log_entry[0])
218
      obj.ops.append(op)
219

    
220
    # Condition to wait for changes
221
    obj.change = threading.Condition(obj.queue._lock)
222

    
223
    return obj
224

    
225
  def Serialize(self):
226
    """Serialize the _JobQueue instance.
227

228
    @rtype: dict
229
    @return: the serialized state
230

231
    """
232
    return {
233
      "id": self.id,
234
      "ops": [op.Serialize() for op in self.ops],
235
      "run_op_index": self.run_op_index,
236
      "start_timestamp": self.start_timestamp,
237
      "end_timestamp": self.end_timestamp,
238
      "received_timestamp": self.received_timestamp,
239
      }
240

    
241
  def CalcStatus(self):
242
    """Compute the status of this job.
243

244
    This function iterates over all the _QueuedOpCodes in the job and
245
    based on their status, computes the job status.
246

247
    The algorithm is:
248
      - if we find a cancelled, or finished with error, the job
249
        status will be the same
250
      - otherwise, the last opcode with the status one of:
251
          - waitlock
252
          - canceling
253
          - running
254

255
        will determine the job status
256

257
      - otherwise, it means either all opcodes are queued, or success,
258
        and the job status will be the same
259

260
    @return: the job status
261

262
    """
263
    status = constants.JOB_STATUS_QUEUED
264

    
265
    all_success = True
266
    for op in self.ops:
267
      if op.status == constants.OP_STATUS_SUCCESS:
268
        continue
269

    
270
      all_success = False
271

    
272
      if op.status == constants.OP_STATUS_QUEUED:
273
        pass
274
      elif op.status == constants.OP_STATUS_WAITLOCK:
275
        status = constants.JOB_STATUS_WAITLOCK
276
      elif op.status == constants.OP_STATUS_RUNNING:
277
        status = constants.JOB_STATUS_RUNNING
278
      elif op.status == constants.OP_STATUS_CANCELING:
279
        status = constants.JOB_STATUS_CANCELING
280
        break
281
      elif op.status == constants.OP_STATUS_ERROR:
282
        status = constants.JOB_STATUS_ERROR
283
        # The whole job fails if one opcode failed
284
        break
285
      elif op.status == constants.OP_STATUS_CANCELED:
286
        status = constants.OP_STATUS_CANCELED
287
        break
288

    
289
    if all_success:
290
      status = constants.JOB_STATUS_SUCCESS
291

    
292
    return status
293

    
294
  def GetLogEntries(self, newer_than):
295
    """Selectively returns the log entries.
296

297
    @type newer_than: None or int
298
    @param newer_than: if this is None, return all log entries,
299
        otherwise return only the log entries with serial higher
300
        than this value
301
    @rtype: list
302
    @return: the list of the log entries selected
303

304
    """
305
    if newer_than is None:
306
      serial = -1
307
    else:
308
      serial = newer_than
309

    
310
    entries = []
311
    for op in self.ops:
312
      entries.extend(filter(lambda entry: entry[0] > serial, op.log))
313

    
314
    return entries
315

    
316
  def MarkUnfinishedOps(self, status, result):
317
    """Mark unfinished opcodes with a given status and result.
318

319
    This is an utility function for marking all running or waiting to
320
    be run opcodes with a given status. Opcodes which are already
321
    finalised are not changed.
322

323
    @param status: a given opcode status
324
    @param result: the opcode result
325

326
    """
327
    not_marked = True
328
    for op in self.ops:
329
      if op.status in constants.OPS_FINALIZED:
330
        assert not_marked, "Finalized opcodes found after non-finalized ones"
331
        continue
332
      op.status = status
333
      op.result = result
334
      not_marked = False
335

    
336

    
337
class _JobQueueWorker(workerpool.BaseWorker):
338
  """The actual job workers.
339

340
  """
341
  def _NotifyStart(self):
342
    """Mark the opcode as running, not lock-waiting.
343

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

349
    """
350
    assert self.queue, "Queue attribute is missing"
351
    assert self.opcode, "Opcode attribute is missing"
352

    
353
    self.queue.acquire()
354
    try:
355
      assert self.opcode.status in (constants.OP_STATUS_WAITLOCK,
356
                                    constants.OP_STATUS_CANCELING)
357

    
358
      # Cancel here if we were asked to
359
      if self.opcode.status == constants.OP_STATUS_CANCELING:
360
        raise CancelJob()
361

    
362
      self.opcode.status = constants.OP_STATUS_RUNNING
363
    finally:
364
      self.queue.release()
365

    
366
  def RunTask(self, job):
367
    """Job executor.
368

369
    This functions processes a job. It is closely tied to the _QueuedJob and
370
    _QueuedOpCode classes.
371

372
    @type job: L{_QueuedJob}
373
    @param job: the job to be processed
374

375
    """
376
    logging.info("Worker %s processing job %s",
377
                  self.worker_id, job.id)
378
    proc = mcpu.Processor(self.pool.queue.context)
379
    self.queue = queue = job.queue
380
    try:
381
      try:
382
        count = len(job.ops)
383
        for idx, op in enumerate(job.ops):
384
          op_summary = op.input.Summary()
385
          if op.status == constants.OP_STATUS_SUCCESS:
386
            # this is a job that was partially completed before master
387
            # daemon shutdown, so it can be expected that some opcodes
388
            # are already completed successfully (if any did error
389
            # out, then the whole job should have been aborted and not
390
            # resubmitted for processing)
391
            logging.info("Op %s/%s: opcode %s already processed, skipping",
392
                         idx + 1, count, op_summary)
393
            continue
394
          try:
395
            logging.info("Op %s/%s: Starting opcode %s", idx + 1, count,
396
                         op_summary)
397

    
398
            queue.acquire()
399
            try:
400
              if op.status == constants.OP_STATUS_CANCELED:
401
                raise CancelJob()
402
              assert op.status == constants.OP_STATUS_QUEUED
403
              job.run_op_index = idx
404
              op.status = constants.OP_STATUS_WAITLOCK
405
              op.result = None
406
              op.start_timestamp = TimeStampNow()
407
              if idx == 0: # first opcode
408
                job.start_timestamp = op.start_timestamp
409
              queue.UpdateJobUnlocked(job)
410

    
411
              input_opcode = op.input
412
            finally:
413
              queue.release()
414

    
415
            def _Log(*args):
416
              """Append a log entry.
417

418
              """
419
              assert len(args) < 3
420

    
421
              if len(args) == 1:
422
                log_type = constants.ELOG_MESSAGE
423
                log_msg = args[0]
424
              else:
425
                log_type, log_msg = args
426

    
427
              # The time is split to make serialization easier and not lose
428
              # precision.
429
              timestamp = utils.SplitTime(time.time())
430

    
431
              queue.acquire()
432
              try:
433
                job.log_serial += 1
434
                op.log.append((job.log_serial, timestamp, log_type, log_msg))
435

    
436
                job.change.notifyAll()
437
              finally:
438
                queue.release()
439

    
440
            # Make sure not to hold lock while _Log is called
441
            self.opcode = op
442
            result = proc.ExecOpCode(input_opcode, _Log, self._NotifyStart)
443

    
444
            queue.acquire()
445
            try:
446
              op.status = constants.OP_STATUS_SUCCESS
447
              op.result = result
448
              op.end_timestamp = TimeStampNow()
449
              queue.UpdateJobUnlocked(job)
450
            finally:
451
              queue.release()
452

    
453
            logging.info("Op %s/%s: Successfully finished opcode %s",
454
                         idx + 1, count, op_summary)
455
          except CancelJob:
456
            # Will be handled further up
457
            raise
458
          except Exception, err:
459
            queue.acquire()
460
            try:
461
              try:
462
                op.status = constants.OP_STATUS_ERROR
463
                if isinstance(err, errors.GenericError):
464
                  op.result = errors.EncodeException(err)
465
                else:
466
                  op.result = str(err)
467
                op.end_timestamp = TimeStampNow()
468
                logging.info("Op %s/%s: Error in opcode %s: %s",
469
                             idx + 1, count, op_summary, err)
470
              finally:
471
                queue.UpdateJobUnlocked(job)
472
            finally:
473
              queue.release()
474
            raise
475

    
476
      except CancelJob:
477
        queue.acquire()
478
        try:
479
          queue.CancelJobUnlocked(job)
480
        finally:
481
          queue.release()
482
      except errors.GenericError, err:
483
        logging.exception("Ganeti exception")
484
      except:
485
        logging.exception("Unhandled exception")
486
    finally:
487
      queue.acquire()
488
      try:
489
        try:
490
          job.run_op_index = -1
491
          job.end_timestamp = TimeStampNow()
492
          queue.UpdateJobUnlocked(job)
493
        finally:
494
          job_id = job.id
495
          status = job.CalcStatus()
496
      finally:
497
        queue.release()
498
      logging.info("Worker %s finished job %s, status = %s",
499
                   self.worker_id, job_id, status)
500

    
501

    
502
class _JobQueueWorkerPool(workerpool.WorkerPool):
503
  """Simple class implementing a job-processing workerpool.
504

505
  """
506
  def __init__(self, queue):
507
    super(_JobQueueWorkerPool, self).__init__(JOBQUEUE_THREADS,
508
                                              _JobQueueWorker)
509
    self.queue = queue
510

    
511

    
512
class JobQueue(object):
513
  """Queue used to manage the jobs.
514

515
  @cvar _RE_JOB_FILE: regex matching the valid job file names
516

517
  """
518
  _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
519

    
520
  def _RequireOpenQueue(fn):
521
    """Decorator for "public" functions.
522

523
    This function should be used for all 'public' functions. That is,
524
    functions usually called from other classes.
525

526
    @warning: Use this decorator only after utils.LockedMethod!
527

528
    Example::
529
      @utils.LockedMethod
530
      @_RequireOpenQueue
531
      def Example(self):
532
        pass
533

534
    """
535
    def wrapper(self, *args, **kwargs):
536
      assert self._queue_lock is not None, "Queue should be open"
537
      return fn(self, *args, **kwargs)
538
    return wrapper
539

    
540
  def __init__(self, context):
541
    """Constructor for JobQueue.
542

543
    The constructor will initialize the job queue object and then
544
    start loading the current jobs from disk, either for starting them
545
    (if they were queue) or for aborting them (if they were already
546
    running).
547

548
    @type context: GanetiContext
549
    @param context: the context object for access to the configuration
550
        data and other ganeti objects
551

552
    """
553
    self.context = context
554
    self._memcache = weakref.WeakValueDictionary()
555
    self._my_hostname = utils.HostInfo().name
556

    
557
    # Locking
558
    self._lock = threading.Lock()
559
    self.acquire = self._lock.acquire
560
    self.release = self._lock.release
561

    
562
    # Initialize
563
    self._queue_lock = jstore.InitAndVerifyQueue(must_lock=True)
564

    
565
    # Read serial file
566
    self._last_serial = jstore.ReadSerial()
567
    assert self._last_serial is not None, ("Serial file was modified between"
568
                                           " check in jstore and here")
569

    
570
    # Get initial list of nodes
571
    self._nodes = dict((n.name, n.primary_ip)
572
                       for n in self.context.cfg.GetAllNodesInfo().values()
573
                       if n.master_candidate)
574

    
575
    # Remove master node
576
    try:
577
      del self._nodes[self._my_hostname]
578
    except KeyError:
579
      pass
580

    
581
    # TODO: Check consistency across nodes
582

    
583
    # Setup worker pool
584
    self._wpool = _JobQueueWorkerPool(self)
585
    try:
586
      # We need to lock here because WorkerPool.AddTask() may start a job while
587
      # we're still doing our work.
588
      self.acquire()
589
      try:
590
        logging.info("Inspecting job queue")
591

    
592
        all_job_ids = self._GetJobIDsUnlocked()
593
        jobs_count = len(all_job_ids)
594
        lastinfo = time.time()
595
        for idx, job_id in enumerate(all_job_ids):
596
          # Give an update every 1000 jobs or 10 seconds
597
          if (idx % 1000 == 0 or time.time() >= (lastinfo + 10.0) or
598
              idx == (jobs_count - 1)):
599
            logging.info("Job queue inspection: %d/%d (%0.1f %%)",
600
                         idx, jobs_count - 1, 100.0 * (idx + 1) / jobs_count)
601
            lastinfo = time.time()
602

    
603
          job = self._LoadJobUnlocked(job_id)
604

    
605
          # a failure in loading the job can cause 'None' to be returned
606
          if job is None:
607
            continue
608

    
609
          status = job.CalcStatus()
610

    
611
          if status in (constants.JOB_STATUS_QUEUED, ):
612
            self._wpool.AddTask(job)
613

    
614
          elif status in (constants.JOB_STATUS_RUNNING,
615
                          constants.JOB_STATUS_WAITLOCK,
616
                          constants.JOB_STATUS_CANCELING):
617
            logging.warning("Unfinished job %s found: %s", job.id, job)
618
            try:
619
              job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,
620
                                    "Unclean master daemon shutdown")
621
            finally:
622
              self.UpdateJobUnlocked(job)
623

    
624
        logging.info("Job queue inspection finished")
625
      finally:
626
        self.release()
627
    except:
628
      self._wpool.TerminateWorkers()
629
      raise
630

    
631
  @utils.LockedMethod
632
  @_RequireOpenQueue
633
  def AddNode(self, node):
634
    """Register a new node with the queue.
635

636
    @type node: L{objects.Node}
637
    @param node: the node object to be added
638

639
    """
640
    node_name = node.name
641
    assert node_name != self._my_hostname
642

    
643
    # Clean queue directory on added node
644
    result = rpc.RpcRunner.call_jobqueue_purge(node_name)
645
    msg = result.RemoteFailMsg()
646
    if msg:
647
      logging.warning("Cannot cleanup queue directory on node %s: %s",
648
                      node_name, msg)
649

    
650
    if not node.master_candidate:
651
      # remove if existing, ignoring errors
652
      self._nodes.pop(node_name, None)
653
      # and skip the replication of the job ids
654
      return
655

    
656
    # Upload the whole queue excluding archived jobs
657
    files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
658

    
659
    # Upload current serial file
660
    files.append(constants.JOB_QUEUE_SERIAL_FILE)
661

    
662
    for file_name in files:
663
      # Read file content
664
      fd = open(file_name, "r")
665
      try:
666
        content = fd.read()
667
      finally:
668
        fd.close()
669

    
670
      result = rpc.RpcRunner.call_jobqueue_update([node_name],
671
                                                  [node.primary_ip],
672
                                                  file_name, content)
673
      msg = result[node_name].RemoteFailMsg()
674
      if msg:
675
        logging.error("Failed to upload file %s to node %s: %s",
676
                      file_name, node_name, msg)
677

    
678
    self._nodes[node_name] = node.primary_ip
679

    
680
  @utils.LockedMethod
681
  @_RequireOpenQueue
682
  def RemoveNode(self, node_name):
683
    """Callback called when removing nodes from the cluster.
684

685
    @type node_name: str
686
    @param node_name: the name of the node to remove
687

688
    """
689
    try:
690
      # The queue is removed by the "leave node" RPC call.
691
      del self._nodes[node_name]
692
    except KeyError:
693
      pass
694

    
695
  def _CheckRpcResult(self, result, nodes, failmsg):
696
    """Verifies the status of an RPC call.
697

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

702
    @param result: the data as returned from the rpc call
703
    @type nodes: list
704
    @param nodes: the list of nodes we made the call to
705
    @type failmsg: str
706
    @param failmsg: the identifier to be used for logging
707

708
    """
709
    failed = []
710
    success = []
711

    
712
    for node in nodes:
713
      msg = result[node].RemoteFailMsg()
714
      if msg:
715
        failed.append(node)
716
        logging.error("RPC call %s failed on node %s: %s",
717
                      result[node].call, node, msg)
718
      else:
719
        success.append(node)
720

    
721
    # +1 for the master node
722
    if (len(success) + 1) < len(failed):
723
      # TODO: Handle failing nodes
724
      logging.error("More than half of the nodes failed")
725

    
726
  def _GetNodeIp(self):
727
    """Helper for returning the node name/ip list.
728

729
    @rtype: (list, list)
730
    @return: a tuple of two lists, the first one with the node
731
        names and the second one with the node addresses
732

733
    """
734
    name_list = self._nodes.keys()
735
    addr_list = [self._nodes[name] for name in name_list]
736
    return name_list, addr_list
737

    
738
  def _WriteAndReplicateFileUnlocked(self, file_name, data):
739
    """Writes a file locally and then replicates it to all nodes.
740

741
    This function will replace the contents of a file on the local
742
    node and then replicate it to all the other nodes we have.
743

744
    @type file_name: str
745
    @param file_name: the path of the file to be replicated
746
    @type data: str
747
    @param data: the new contents of the file
748

749
    """
750
    utils.WriteFile(file_name, data=data)
751

    
752
    names, addrs = self._GetNodeIp()
753
    result = rpc.RpcRunner.call_jobqueue_update(names, addrs, file_name, data)
754
    self._CheckRpcResult(result, self._nodes,
755
                         "Updating %s" % file_name)
756

    
757
  def _RenameFilesUnlocked(self, rename):
758
    """Renames a file locally and then replicate the change.
759

760
    This function will rename a file in the local queue directory
761
    and then replicate this rename to all the other nodes we have.
762

763
    @type rename: list of (old, new)
764
    @param rename: List containing tuples mapping old to new names
765

766
    """
767
    # Rename them locally
768
    for old, new in rename:
769
      utils.RenameFile(old, new, mkdir=True)
770

    
771
    # ... and on all nodes
772
    names, addrs = self._GetNodeIp()
773
    result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
774
    self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
775

    
776
  def _FormatJobID(self, job_id):
777
    """Convert a job ID to string format.
778

779
    Currently this just does C{str(job_id)} after performing some
780
    checks, but if we want to change the job id format this will
781
    abstract this change.
782

783
    @type job_id: int or long
784
    @param job_id: the numeric job id
785
    @rtype: str
786
    @return: the formatted job id
787

788
    """
789
    if not isinstance(job_id, (int, long)):
790
      raise errors.ProgrammerError("Job ID '%s' not numeric" % job_id)
791
    if job_id < 0:
792
      raise errors.ProgrammerError("Job ID %s is negative" % job_id)
793

    
794
    return str(job_id)
795

    
796
  @classmethod
797
  def _GetArchiveDirectory(cls, job_id):
798
    """Returns the archive directory for a job.
799

800
    @type job_id: str
801
    @param job_id: Job identifier
802
    @rtype: str
803
    @return: Directory name
804

805
    """
806
    return str(int(job_id) / JOBS_PER_ARCHIVE_DIRECTORY)
807

    
808
  def _NewSerialUnlocked(self):
809
    """Generates a new job identifier.
810

811
    Job identifiers are unique during the lifetime of a cluster.
812

813
    @rtype: str
814
    @return: a string representing the job identifier.
815

816
    """
817
    # New number
818
    serial = self._last_serial + 1
819

    
820
    # Write to file
821
    self._WriteAndReplicateFileUnlocked(constants.JOB_QUEUE_SERIAL_FILE,
822
                                        "%s\n" % serial)
823

    
824
    # Keep it only if we were able to write the file
825
    self._last_serial = serial
826

    
827
    return self._FormatJobID(serial)
828

    
829
  @staticmethod
830
  def _GetJobPath(job_id):
831
    """Returns the job file for a given job id.
832

833
    @type job_id: str
834
    @param job_id: the job identifier
835
    @rtype: str
836
    @return: the path to the job file
837

838
    """
839
    return os.path.join(constants.QUEUE_DIR, "job-%s" % job_id)
840

    
841
  @classmethod
842
  def _GetArchivedJobPath(cls, job_id):
843
    """Returns the archived job file for a give job id.
844

845
    @type job_id: str
846
    @param job_id: the job identifier
847
    @rtype: str
848
    @return: the path to the archived job file
849

850
    """
851
    path = "%s/job-%s" % (cls._GetArchiveDirectory(job_id), job_id)
852
    return os.path.join(constants.JOB_QUEUE_ARCHIVE_DIR, path)
853

    
854
  @classmethod
855
  def _ExtractJobID(cls, name):
856
    """Extract the job id from a filename.
857

858
    @type name: str
859
    @param name: the job filename
860
    @rtype: job id or None
861
    @return: the job id corresponding to the given filename,
862
        or None if the filename does not represent a valid
863
        job file
864

865
    """
866
    m = cls._RE_JOB_FILE.match(name)
867
    if m:
868
      return m.group(1)
869
    else:
870
      return None
871

    
872
  def _GetJobIDsUnlocked(self, archived=False):
873
    """Return all known job IDs.
874

875
    If the parameter archived is True, archived jobs IDs will be
876
    included. Currently this argument is unused.
877

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

882
    @rtype: list
883
    @return: the list of job IDs
884

885
    """
886
    jlist = [self._ExtractJobID(name) for name in self._ListJobFiles()]
887
    jlist = utils.NiceSort(jlist)
888
    return jlist
889

    
890
  def _ListJobFiles(self):
891
    """Returns the list of current job files.
892

893
    @rtype: list
894
    @return: the list of job file names
895

896
    """
897
    return [name for name in utils.ListVisibleFiles(constants.QUEUE_DIR)
898
            if self._RE_JOB_FILE.match(name)]
899

    
900
  def _LoadJobUnlocked(self, job_id):
901
    """Loads a job from the disk or memory.
902

903
    Given a job id, this will return the cached job object if
904
    existing, or try to load the job from the disk. If loading from
905
    disk, it will also add the job to the cache.
906

907
    @param job_id: the job id
908
    @rtype: L{_QueuedJob} or None
909
    @return: either None or the job object
910

911
    """
912
    job = self._memcache.get(job_id, None)
913
    if job:
914
      logging.debug("Found job %s in memcache", job_id)
915
      return job
916

    
917
    filepath = self._GetJobPath(job_id)
918
    logging.debug("Loading job from %s", filepath)
919
    try:
920
      fd = open(filepath, "r")
921
    except IOError, err:
922
      if err.errno in (errno.ENOENT, ):
923
        return None
924
      raise
925
    try:
926
      data = serializer.LoadJson(fd.read())
927
    finally:
928
      fd.close()
929

    
930
    try:
931
      job = _QueuedJob.Restore(self, data)
932
    except Exception, err:
933
      new_path = self._GetArchivedJobPath(job_id)
934
      if filepath == new_path:
935
        # job already archived (future case)
936
        logging.exception("Can't parse job %s", job_id)
937
      else:
938
        # non-archived case
939
        logging.exception("Can't parse job %s, will archive.", job_id)
940
        self._RenameFilesUnlocked([(filepath, new_path)])
941
      return None
942

    
943
    self._memcache[job_id] = job
944
    logging.debug("Added job %s to the cache", job_id)
945
    return job
946

    
947
  def _GetJobsUnlocked(self, job_ids):
948
    """Return a list of jobs based on their IDs.
949

950
    @type job_ids: list
951
    @param job_ids: either an empty list (meaning all jobs),
952
        or a list of job IDs
953
    @rtype: list
954
    @return: the list of job objects
955

956
    """
957
    if not job_ids:
958
      job_ids = self._GetJobIDsUnlocked()
959

    
960
    return [self._LoadJobUnlocked(job_id) for job_id in job_ids]
961

    
962
  @staticmethod
963
  def _IsQueueMarkedDrain():
964
    """Check if the queue is marked from drain.
965

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

969
    @rtype: boolean
970
    @return: True of the job queue is marked for draining
971

972
    """
973
    return os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
974

    
975
  @staticmethod
976
  def SetDrainFlag(drain_flag):
977
    """Sets the drain flag for the queue.
978

979
    This is similar to the function L{backend.JobQueueSetDrainFlag},
980
    and in the future we might merge them.
981

982
    @type drain_flag: boolean
983
    @param drain_flag: Whether to set or unset the drain flag
984

985
    """
986
    if drain_flag:
987
      utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
988
    else:
989
      utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
990
    return True
991

    
992
  @_RequireOpenQueue
993
  def _SubmitJobUnlocked(self, ops):
994
    """Create and store a new job.
995

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

999
    @type ops: list
1000
    @param ops: The list of OpCodes that will become the new job.
1001
    @rtype: job ID
1002
    @return: the job ID of the newly created job
1003
    @raise errors.JobQueueDrainError: if the job is marked for draining
1004

1005
    """
1006
    if self._IsQueueMarkedDrain():
1007
      raise errors.JobQueueDrainError("Job queue is drained, refusing job")
1008

    
1009
    # Check job queue size
1010
    size = len(self._ListJobFiles())
1011
    if size >= constants.JOB_QUEUE_SIZE_SOFT_LIMIT:
1012
      # TODO: Autoarchive jobs. Make sure it's not done on every job
1013
      # submission, though.
1014
      #size = ...
1015
      pass
1016

    
1017
    if size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
1018
      raise errors.JobQueueFull()
1019

    
1020
    # Get job identifier
1021
    job_id = self._NewSerialUnlocked()
1022
    job = _QueuedJob(self, job_id, ops)
1023

    
1024
    # Write to disk
1025
    self.UpdateJobUnlocked(job)
1026

    
1027
    logging.debug("Adding new job %s to the cache", job_id)
1028
    self._memcache[job_id] = job
1029

    
1030
    # Add to worker pool
1031
    self._wpool.AddTask(job)
1032

    
1033
    return job.id
1034

    
1035
  @utils.LockedMethod
1036
  @_RequireOpenQueue
1037
  def SubmitJob(self, ops):
1038
    """Create and store a new job.
1039

1040
    @see: L{_SubmitJobUnlocked}
1041

1042
    """
1043
    return self._SubmitJobUnlocked(ops)
1044

    
1045
  @utils.LockedMethod
1046
  @_RequireOpenQueue
1047
  def SubmitManyJobs(self, jobs):
1048
    """Create and store multiple jobs.
1049

1050
    @see: L{_SubmitJobUnlocked}
1051

1052
    """
1053
    results = []
1054
    for ops in jobs:
1055
      try:
1056
        data = self._SubmitJobUnlocked(ops)
1057
        status = True
1058
      except errors.GenericError, err:
1059
        data = str(err)
1060
        status = False
1061
      results.append((status, data))
1062

    
1063
    return results
1064

    
1065

    
1066
  @_RequireOpenQueue
1067
  def UpdateJobUnlocked(self, job):
1068
    """Update a job's on disk storage.
1069

1070
    After a job has been modified, this function needs to be called in
1071
    order to write the changes to disk and replicate them to the other
1072
    nodes.
1073

1074
    @type job: L{_QueuedJob}
1075
    @param job: the changed job
1076

1077
    """
1078
    filename = self._GetJobPath(job.id)
1079
    data = serializer.DumpJson(job.Serialize(), indent=False)
1080
    logging.debug("Writing job %s to %s", job.id, filename)
1081
    self._WriteAndReplicateFileUnlocked(filename, data)
1082

    
1083
    # Notify waiters about potential changes
1084
    job.change.notifyAll()
1085

    
1086
  @utils.LockedMethod
1087
  @_RequireOpenQueue
1088
  def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial,
1089
                        timeout):
1090
    """Waits for changes in a job.
1091

1092
    @type job_id: string
1093
    @param job_id: Job identifier
1094
    @type fields: list of strings
1095
    @param fields: Which fields to check for changes
1096
    @type prev_job_info: list or None
1097
    @param prev_job_info: Last job information returned
1098
    @type prev_log_serial: int
1099
    @param prev_log_serial: Last job message serial number
1100
    @type timeout: float
1101
    @param timeout: maximum time to wait
1102
    @rtype: tuple (job info, log entries)
1103
    @return: a tuple of the job information as required via
1104
        the fields parameter, and the log entries as a list
1105

1106
        if the job has not changed and the timeout has expired,
1107
        we instead return a special value,
1108
        L{constants.JOB_NOTCHANGED}, which should be interpreted
1109
        as such by the clients
1110

1111
    """
1112
    logging.debug("Waiting for changes in job %s", job_id)
1113

    
1114
    job_info = None
1115
    log_entries = None
1116

    
1117
    end_time = time.time() + timeout
1118
    while True:
1119
      delta_time = end_time - time.time()
1120
      if delta_time < 0:
1121
        return constants.JOB_NOTCHANGED
1122

    
1123
      job = self._LoadJobUnlocked(job_id)
1124
      if not job:
1125
        logging.debug("Job %s not found", job_id)
1126
        break
1127

    
1128
      status = job.CalcStatus()
1129
      job_info = self._GetJobInfoUnlocked(job, fields)
1130
      log_entries = job.GetLogEntries(prev_log_serial)
1131

    
1132
      # Serializing and deserializing data can cause type changes (e.g. from
1133
      # tuple to list) or precision loss. We're doing it here so that we get
1134
      # the same modifications as the data received from the client. Without
1135
      # this, the comparison afterwards might fail without the data being
1136
      # significantly different.
1137
      job_info = serializer.LoadJson(serializer.DumpJson(job_info))
1138
      log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
1139

    
1140
      if status not in (constants.JOB_STATUS_QUEUED,
1141
                        constants.JOB_STATUS_RUNNING,
1142
                        constants.JOB_STATUS_WAITLOCK):
1143
        # Don't even try to wait if the job is no longer running, there will be
1144
        # no changes.
1145
        break
1146

    
1147
      if (prev_job_info != job_info or
1148
          (log_entries and prev_log_serial != log_entries[0][0])):
1149
        break
1150

    
1151
      logging.debug("Waiting again")
1152

    
1153
      # Release the queue lock while waiting
1154
      job.change.wait(delta_time)
1155

    
1156
    logging.debug("Job %s changed", job_id)
1157

    
1158
    if job_info is None and log_entries is None:
1159
      return None
1160
    else:
1161
      return (job_info, log_entries)
1162

    
1163
  @utils.LockedMethod
1164
  @_RequireOpenQueue
1165
  def CancelJob(self, job_id):
1166
    """Cancels a job.
1167

1168
    This will only succeed if the job has not started yet.
1169

1170
    @type job_id: string
1171
    @param job_id: job ID of job to be cancelled.
1172

1173
    """
1174
    logging.info("Cancelling job %s", job_id)
1175

    
1176
    job = self._LoadJobUnlocked(job_id)
1177
    if not job:
1178
      logging.debug("Job %s not found", job_id)
1179
      return (False, "Job %s not found" % job_id)
1180

    
1181
    job_status = job.CalcStatus()
1182

    
1183
    if job_status not in (constants.JOB_STATUS_QUEUED,
1184
                          constants.JOB_STATUS_WAITLOCK):
1185
      logging.debug("Job %s is no longer waiting in the queue", job.id)
1186
      return (False, "Job %s is no longer waiting in the queue" % job.id)
1187

    
1188
    if job_status == constants.JOB_STATUS_QUEUED:
1189
      self.CancelJobUnlocked(job)
1190
      return (True, "Job %s canceled" % job.id)
1191

    
1192
    elif job_status == constants.JOB_STATUS_WAITLOCK:
1193
      # The worker will notice the new status and cancel the job
1194
      try:
1195
        job.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
1196
      finally:
1197
        self.UpdateJobUnlocked(job)
1198
      return (True, "Job %s will be canceled" % job.id)
1199

    
1200
  @_RequireOpenQueue
1201
  def CancelJobUnlocked(self, job):
1202
    """Marks a job as canceled.
1203

1204
    """
1205
    try:
1206
      job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
1207
                            "Job canceled by request")
1208
    finally:
1209
      self.UpdateJobUnlocked(job)
1210

    
1211
  @_RequireOpenQueue
1212
  def _ArchiveJobsUnlocked(self, jobs):
1213
    """Archives jobs.
1214

1215
    @type jobs: list of L{_QueuedJob}
1216
    @param jobs: Job objects
1217
    @rtype: int
1218
    @return: Number of archived jobs
1219

1220
    """
1221
    archive_jobs = []
1222
    rename_files = []
1223
    for job in jobs:
1224
      if job.CalcStatus() not in (constants.JOB_STATUS_CANCELED,
1225
                                  constants.JOB_STATUS_SUCCESS,
1226
                                  constants.JOB_STATUS_ERROR):
1227
        logging.debug("Job %s is not yet done", job.id)
1228
        continue
1229

    
1230
      archive_jobs.append(job)
1231

    
1232
      old = self._GetJobPath(job.id)
1233
      new = self._GetArchivedJobPath(job.id)
1234
      rename_files.append((old, new))
1235

    
1236
    # TODO: What if 1..n files fail to rename?
1237
    self._RenameFilesUnlocked(rename_files)
1238

    
1239
    logging.debug("Successfully archived job(s) %s",
1240
                  ", ".join(job.id for job in archive_jobs))
1241

    
1242
    return len(archive_jobs)
1243

    
1244
  @utils.LockedMethod
1245
  @_RequireOpenQueue
1246
  def ArchiveJob(self, job_id):
1247
    """Archives a job.
1248

1249
    This is just a wrapper over L{_ArchiveJobsUnlocked}.
1250

1251
    @type job_id: string
1252
    @param job_id: Job ID of job to be archived.
1253
    @rtype: bool
1254
    @return: Whether job was archived
1255

1256
    """
1257
    logging.info("Archiving job %s", job_id)
1258

    
1259
    job = self._LoadJobUnlocked(job_id)
1260
    if not job:
1261
      logging.debug("Job %s not found", job_id)
1262
      return False
1263

    
1264
    return self._ArchiveJobsUnlocked([job]) == 1
1265

    
1266
  @utils.LockedMethod
1267
  @_RequireOpenQueue
1268
  def AutoArchiveJobs(self, age, timeout):
1269
    """Archives all jobs based on age.
1270

1271
    The method will archive all jobs which are older than the age
1272
    parameter. For jobs that don't have an end timestamp, the start
1273
    timestamp will be considered. The special '-1' age will cause
1274
    archival of all jobs (that are not running or queued).
1275

1276
    @type age: int
1277
    @param age: the minimum age in seconds
1278

1279
    """
1280
    logging.info("Archiving jobs with age more than %s seconds", age)
1281

    
1282
    now = time.time()
1283
    end_time = now + timeout
1284
    archived_count = 0
1285
    last_touched = 0
1286

    
1287
    all_job_ids = self._GetJobIDsUnlocked(archived=False)
1288
    pending = []
1289
    for idx, job_id in enumerate(all_job_ids):
1290
      last_touched = idx
1291

    
1292
      # Not optimal because jobs could be pending
1293
      # TODO: Measure average duration for job archival and take number of
1294
      # pending jobs into account.
1295
      if time.time() > end_time:
1296
        break
1297

    
1298
      # Returns None if the job failed to load
1299
      job = self._LoadJobUnlocked(job_id)
1300
      if job:
1301
        if job.end_timestamp is None:
1302
          if job.start_timestamp is None:
1303
            job_age = job.received_timestamp
1304
          else:
1305
            job_age = job.start_timestamp
1306
        else:
1307
          job_age = job.end_timestamp
1308

    
1309
        if age == -1 or now - job_age[0] > age:
1310
          pending.append(job)
1311

    
1312
          # Archive 10 jobs at a time
1313
          if len(pending) >= 10:
1314
            archived_count += self._ArchiveJobsUnlocked(pending)
1315
            pending = []
1316

    
1317
    if pending:
1318
      archived_count += self._ArchiveJobsUnlocked(pending)
1319

    
1320
    return (archived_count, len(all_job_ids) - last_touched - 1)
1321

    
1322
  def _GetJobInfoUnlocked(self, job, fields):
1323
    """Returns information about a job.
1324

1325
    @type job: L{_QueuedJob}
1326
    @param job: the job which we query
1327
    @type fields: list
1328
    @param fields: names of fields to return
1329
    @rtype: list
1330
    @return: list with one element for each field
1331
    @raise errors.OpExecError: when an invalid field
1332
        has been passed
1333

1334
    """
1335
    row = []
1336
    for fname in fields:
1337
      if fname == "id":
1338
        row.append(job.id)
1339
      elif fname == "status":
1340
        row.append(job.CalcStatus())
1341
      elif fname == "ops":
1342
        row.append([op.input.__getstate__() for op in job.ops])
1343
      elif fname == "opresult":
1344
        row.append([op.result for op in job.ops])
1345
      elif fname == "opstatus":
1346
        row.append([op.status for op in job.ops])
1347
      elif fname == "oplog":
1348
        row.append([op.log for op in job.ops])
1349
      elif fname == "opstart":
1350
        row.append([op.start_timestamp for op in job.ops])
1351
      elif fname == "opend":
1352
        row.append([op.end_timestamp for op in job.ops])
1353
      elif fname == "received_ts":
1354
        row.append(job.received_timestamp)
1355
      elif fname == "start_ts":
1356
        row.append(job.start_timestamp)
1357
      elif fname == "end_ts":
1358
        row.append(job.end_timestamp)
1359
      elif fname == "summary":
1360
        row.append([op.input.Summary() for op in job.ops])
1361
      else:
1362
        raise errors.OpExecError("Invalid job query field '%s'" % fname)
1363
    return row
1364

    
1365
  @utils.LockedMethod
1366
  @_RequireOpenQueue
1367
  def QueryJobs(self, job_ids, fields):
1368
    """Returns a list of jobs in queue.
1369

1370
    This is a wrapper of L{_GetJobsUnlocked}, which actually does the
1371
    processing for each job.
1372

1373
    @type job_ids: list
1374
    @param job_ids: sequence of job identifiers or None for all
1375
    @type fields: list
1376
    @param fields: names of fields to return
1377
    @rtype: list
1378
    @return: list one element per job, each element being list with
1379
        the requested fields
1380

1381
    """
1382
    jobs = []
1383

    
1384
    for job in self._GetJobsUnlocked(job_ids):
1385
      if job is None:
1386
        jobs.append(None)
1387
      else:
1388
        jobs.append(self._GetJobInfoUnlocked(job, fields))
1389

    
1390
    return jobs
1391

    
1392
  @utils.LockedMethod
1393
  @_RequireOpenQueue
1394
  def Shutdown(self):
1395
    """Stops the job queue.
1396

1397
    This shutdowns all the worker threads an closes the queue.
1398

1399
    """
1400
    self._wpool.TerminateWorkers()
1401

    
1402
    self._queue_lock.Close()
1403
    self._queue_lock = None