Source Code for Module ganeti.jstore

  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() 
171