code.grnet.gr Git - ganeti-local/blob - lib/jstore.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007 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 import errno
  25 import os
  26
  27 from ganeti import constants
  28 from ganeti import errors
  29 from ganeti import runtime
  30 from ganeti import utils
  31
  32
  33 def _ReadNumericFile(file_name):
  34   """Reads a file containing a number.
  35
  36   @rtype: None or int
  37   @return: None if file is not found, otherwise number
  38
  39   """
  40   try:
  41     return int(utils.ReadFile(file_name))
  42   except EnvironmentError, err:
  43     if err.errno in (errno.ENOENT, ):
  44       return None
  45     raise
  46
  47
  48 def ReadSerial():
  49   """Read the serial file.
  50
  51   The queue should be locked while this function is called.
  52
  53   """
  54   return _ReadNumericFile(constants.JOB_QUEUE_SERIAL_FILE)
  55
  56
  57 def ReadVersion():
  58   """Read the queue version.
  59
  60   The queue should be locked while this function is called.
  61
  62   """
  63   return _ReadNumericFile(constants.JOB_QUEUE_VERSION_FILE)
  64
  65
  66 def InitAndVerifyQueue(must_lock):
  67   """Open and lock job queue.
  68
  69   If necessary, the queue is automatically initialized.
  70
  71   @type must_lock: bool
  72   @param must_lock: Whether an exclusive lock must be held.
  73   @rtype: utils.FileLock
  74   @return: Lock object for the queue. This can be used to change the
  75            locking mode.
  76
  77   """
  78   getents = runtime.GetEnts()
  79
  80   # Lock queue
  81   queue_lock = utils.FileLock.Open(constants.JOB_QUEUE_LOCK_FILE)
  82   try:
  83     # The queue needs to be locked in exclusive mode to write to the serial and
  84     # version files.
  85     if must_lock:
  86       queue_lock.Exclusive(blocking=True)
  87       holding_lock = True
  88     else:
  89       try:
  90         queue_lock.Exclusive(blocking=False)
  91         holding_lock = True
  92       except errors.LockError:
  93         # Ignore errors and assume the process keeping the lock checked
  94         # everything.
  95         holding_lock = False
  96
  97     if holding_lock:
  98       # Verify version
  99       version = ReadVersion()
 100       if version is None:
 101         # Write new version file
 102         utils.WriteFile(constants.JOB_QUEUE_VERSION_FILE,
 103                         uid=getents.masterd_uid, gid=getents.masterd_gid,
 104                         data="%s\n" % constants.JOB_QUEUE_VERSION)
 105
 106         # Read again
 107         version = ReadVersion()
 108
 109       if version != constants.JOB_QUEUE_VERSION:
 110         raise errors.JobQueueError("Found job queue version %s, expected %s",
 111                                    version, constants.JOB_QUEUE_VERSION)
 112
 113       serial = ReadSerial()
 114       if serial is None:
 115         # Write new serial file
 116         utils.WriteFile(constants.JOB_QUEUE_SERIAL_FILE,
 117                         uid=getents.masterd_uid, gid=getents.masterd_gid,
 118                         data="%s\n" % 0)
 119
 120         # Read again
 121         serial = ReadSerial()
 122
 123       if serial is None:
 124         # There must be a serious problem
 125         raise errors.JobQueueError("Can't read/parse the job queue"
 126                                    " serial file")
 127
 128       if not must_lock:
 129         # There's no need for more error handling. Closing the lock
 130         # file below in case of an error will unlock it anyway.
 131         queue_lock.Unlock()
 132
 133   except:
 134     queue_lock.Close()
 135     raise
 136
 137   return queue_lock
 138
 139
 140 def CheckDrainFlag():
 141   """Check if the queue is marked to be drained.
 142
 143   This currently uses the queue drain file, which makes it a per-node flag.
 144   In the future this can be moved to the config file.
 145
 146   @rtype: boolean
 147   @return: True if the job queue is marked drained
 148
 149   """
 150   return os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
 151
 152
 153 def SetDrainFlag(drain_flag):
 154   """Sets the drain flag for the queue.
 155
 156   @type drain_flag: boolean
 157   @param drain_flag: Whether to set or unset the drain flag
 158   @attention: This function should only called the current holder of the queue
 159     lock
 160
 161   """
 162   getents = runtime.GetEnts()
 163
 164   if drain_flag:
 165     utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="",
 166                     uid=getents.masterd_uid, gid=getents.masterd_gid)
 167   else:
 168     utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
 169
 170   assert (not drain_flag) ^ CheckDrainFlag()