code.grnet.gr Git - ganeti-local/blob - daemons/ganeti-watcher

   1 #!/usr/bin/python
   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 """Tool to restart erronously downed virtual machines.
  23
  24 This program and set of classes implement a watchdog to restart
  25 virtual machines in a Ganeti cluster that have crashed or been killed
  26 by a node reboot.  Run from cron or similar.
  27
  28 """
  29
  30 import os
  31 import sys
  32 import re
  33 import time
  34 import fcntl
  35 import errno
  36 import logging
  37 from optparse import OptionParser
  38
  39 from ganeti import utils
  40 from ganeti import constants
  41 from ganeti import serializer
  42 from ganeti import ssconf
  43 from ganeti import errors
  44
  45
  46 MAXTRIES = 5
  47 BAD_STATES = ['stopped']
  48 HELPLESS_STATES = ['(node down)']
  49 NOTICE = 'NOTICE'
  50 ERROR = 'ERROR'
  51 KEY_RESTART_COUNT = "restart_count"
  52 KEY_RESTART_WHEN = "restart_when"
  53 KEY_BOOT_ID = "bootid"
  54
  55
  56 class Error(Exception):
  57   """Generic custom error class."""
  58
  59
  60 class NotMasterError(Error):
  61   """Exception raised when this host is not the master."""
  62
  63
  64 def Indent(s, prefix='| '):
  65   """Indent a piece of text with a given prefix before each line.
  66
  67   Args:
  68     s: The string to indent
  69     prefix: The string to prepend each line.
  70
  71   """
  72   return "%s%s\n" % (prefix, ('\n' + prefix).join(s.splitlines()))
  73
  74
  75 def DoCmd(cmd):
  76   """Run a shell command.
  77
  78   Args:
  79     cmd: the command to run.
  80
  81   Raises CommandError with verbose commentary on error.
  82
  83   """
  84   res = utils.RunCmd(cmd)
  85
  86   if res.failed:
  87     raise Error("Command %s failed:\n%s\nstdout:\n%sstderr:\n%s" %
  88                 (repr(cmd),
  89                  Indent(res.fail_reason),
  90                  Indent(res.stdout),
  91                  Indent(res.stderr)))
  92
  93   return res
  94
  95
  96 def LockFile(fd):
  97   """Locks a file using POSIX locks.
  98
  99   """
 100   try:
 101     fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
 102   except IOError, err:
 103     if err.errno == errno.EAGAIN:
 104       raise StandardError("File already locked")
 105     raise
 106
 107
 108 class WatcherState(object):
 109   """Interface to a state file recording restart attempts.
 110
 111   """
 112   def __init__(self):
 113     """Open, lock, read and parse the file.
 114
 115     Raises StandardError on lock contention.
 116
 117     """
 118     # The two-step dance below is necessary to allow both opening existing
 119     # file read/write and creating if not existing.  Vanilla open will truncate
 120     # an existing file -or- allow creating if not existing.
 121     fd = os.open(constants.WATCHER_STATEFILE, os.O_RDWR | os.O_CREAT)
 122     self.statefile = os.fdopen(fd, 'w+')
 123
 124     LockFile(self.statefile.fileno())
 125
 126     try:
 127       self.data = serializer.Load(self.statefile.read())
 128     except Exception, msg:
 129       # Ignore errors while loading the file and treat it as empty
 130       self.data = {}
 131       logging.warning(("Empty or invalid state file. Using defaults."
 132                        " Error message: %s"), msg)
 133
 134     if "instance" not in self.data:
 135       self.data["instance"] = {}
 136     if "node" not in self.data:
 137       self.data["node"] = {}
 138
 139   def Save(self):
 140     """Save state to file, then unlock and close it.
 141
 142     """
 143     assert self.statefile
 144
 145     # We need to make sure the file is locked before renaming it, otherwise
 146     # starting ganeti-watcher again at the same time will create a conflict.
 147     fd = utils.WriteFile(constants.WATCHER_STATEFILE,
 148                          data=serializer.Dump(self.data),
 149                          prewrite=LockFile, close=False)
 150     self.statefile = os.fdopen(fd, 'w+')
 151
 152   def Close(self):
 153     """Unlock configuration file and close it.
 154
 155     """
 156     assert self.statefile
 157
 158     # Files are automatically unlocked when closing them
 159     self.statefile.close()
 160     self.statefile = None
 161
 162   def GetNodeBootID(self, name):
 163     """Returns the last boot ID of a node or None.
 164
 165     """
 166     ndata = self.data["node"]
 167
 168     if name in ndata and KEY_BOOT_ID in ndata[name]:
 169       return ndata[name][KEY_BOOT_ID]
 170     return None
 171
 172   def SetNodeBootID(self, name, bootid):
 173     """Sets the boot ID of a node.
 174
 175     """
 176     assert bootid
 177
 178     ndata = self.data["node"]
 179
 180     if name not in ndata:
 181       ndata[name] = {}
 182
 183     ndata[name][KEY_BOOT_ID] = bootid
 184
 185   def NumberOfRestartAttempts(self, instance):
 186     """Returns number of previous restart attempts.
 187
 188     Args:
 189       instance - the instance to look up.
 190
 191     """
 192     idata = self.data["instance"]
 193
 194     if instance.name in idata:
 195       return idata[instance.name][KEY_RESTART_COUNT]
 196
 197     return 0
 198
 199   def RecordRestartAttempt(self, instance):
 200     """Record a restart attempt.
 201
 202     Args:
 203       instance - the instance being restarted
 204
 205     """
 206     idata = self.data["instance"]
 207
 208     if instance.name not in idata:
 209       inst = idata[instance.name] = {}
 210     else:
 211       inst = idata[instance.name]
 212
 213     inst[KEY_RESTART_WHEN] = time.time()
 214     inst[KEY_RESTART_COUNT] = inst.get(KEY_RESTART_COUNT, 0) + 1
 215
 216   def RemoveInstance(self, instance):
 217     """Update state to reflect that a machine is running, i.e. remove record.
 218
 219     Args:
 220       instance - the instance to remove from books
 221
 222     This method removes the record for a named instance.
 223
 224     """
 225     idata = self.data["instance"]
 226
 227     if instance.name in idata:
 228       del idata[instance.name]
 229
 230
 231 class Instance(object):
 232   """Abstraction for a Virtual Machine instance.
 233
 234   Methods:
 235     Restart(): issue a command to restart the represented machine.
 236
 237   """
 238   def __init__(self, name, state, autostart):
 239     self.name = name
 240     self.state = state
 241     self.autostart = autostart
 242
 243   def Restart(self):
 244     """Encapsulates the start of an instance.
 245
 246     """
 247     DoCmd(['gnt-instance', 'startup', '--lock-retries=15', self.name])
 248
 249   def ActivateDisks(self):
 250     """Encapsulates the activation of all disks of an instance.
 251
 252     """
 253     DoCmd(['gnt-instance', 'activate-disks', '--lock-retries=15', self.name])
 254
 255
 256 def _RunListCmd(cmd):
 257   """Runs a command and parses its output into lists.
 258
 259   """
 260   for line in DoCmd(cmd).stdout.splitlines():
 261     yield line.split(':')
 262
 263
 264 def GetInstanceList(with_secondaries=None):
 265   """Get a list of instances on this cluster.
 266
 267   """
 268   cmd = ['gnt-instance', 'list', '--lock-retries=15', '--no-headers',
 269          '--separator=:']
 270
 271   fields = 'name,oper_state,admin_state'
 272
 273   if with_secondaries is not None:
 274     fields += ',snodes'
 275
 276   cmd.append('-o')
 277   cmd.append(fields)
 278
 279   instances = []
 280   for fields in _RunListCmd(cmd):
 281     if with_secondaries is not None:
 282       (name, status, autostart, snodes) = fields
 283
 284       if snodes == "-":
 285         continue
 286
 287       for node in with_secondaries:
 288         if node in snodes.split(','):
 289           break
 290       else:
 291         continue
 292
 293     else:
 294       (name, status, autostart) = fields
 295
 296     instances.append(Instance(name, status, autostart != "no"))
 297
 298   return instances
 299
 300
 301 def GetNodeBootIDs():
 302   """Get a dict mapping nodes to boot IDs.
 303
 304   """
 305   cmd = ['gnt-node', 'list', '--lock-retries=15', '--no-headers',
 306          '--separator=:', '-o', 'name,bootid']
 307
 308   ids = {}
 309   for fields in _RunListCmd(cmd):
 310     (name, bootid) = fields
 311     ids[name] = bootid
 312
 313   return ids
 314
 315
 316 class Watcher(object):
 317   """Encapsulate the logic for restarting erronously halted virtual machines.
 318
 319   The calling program should periodically instantiate me and call Run().
 320   This will traverse the list of instances, and make up to MAXTRIES attempts
 321   to restart machines that are down.
 322
 323   """
 324   def __init__(self):
 325     sstore = ssconf.SimpleStore()
 326     master = sstore.GetMasterNode()
 327     if master != utils.HostInfo().name:
 328       raise NotMasterError("This is not the master node")
 329     self.instances = GetInstanceList()
 330     self.bootids = GetNodeBootIDs()
 331     self.started_instances = set()
 332
 333   def Run(self):
 334     notepad = WatcherState()
 335     try:
 336       self.CheckInstances(notepad)
 337       self.CheckDisks(notepad)
 338       self.VerifyDisks()
 339     finally:
 340       notepad.Save()
 341
 342   def CheckDisks(self, notepad):
 343     """Check all nodes for restarted ones.
 344
 345     """
 346     check_nodes = []
 347     for name, id in self.bootids.iteritems():
 348       old = notepad.GetNodeBootID(name)
 349       if old != id:
 350         # Node's boot ID has changed, proably through a reboot.
 351         check_nodes.append(name)
 352
 353     if check_nodes:
 354       # Activate disks for all instances with any of the checked nodes as a
 355       # secondary node.
 356       for instance in GetInstanceList(with_secondaries=check_nodes):
 357         if not instance.autostart:
 358           logging.info(("Skipping disk activation for non-autostart"
 359                         " instance %s"), instance.name)
 360           continue
 361         if instance.name in self.started_instances:
 362           # we already tried to start the instance, which should have
 363           # activated its drives (if they can be at all)
 364           continue
 365         try:
 366           logging.info("Activating disks for instance %s", instance.name)
 367           instance.ActivateDisks()
 368         except Error, err:
 369           logging.error(str(err), exc_info=True)
 370
 371       # Keep changed boot IDs
 372       for name in check_nodes:
 373         notepad.SetNodeBootID(name, self.bootids[name])
 374
 375   def CheckInstances(self, notepad):
 376     """Make a pass over the list of instances, restarting downed ones.
 377
 378     """
 379     for instance in self.instances:
 380       # Don't care about manually stopped instances
 381       if not instance.autostart:
 382         continue
 383
 384       if instance.state in BAD_STATES:
 385         n = notepad.NumberOfRestartAttempts(instance)
 386
 387         if n > MAXTRIES:
 388           # stay quiet.
 389           continue
 390         elif n < MAXTRIES:
 391           last = " (Attempt #%d)" % (n + 1)
 392         else:
 393           notepad.RecordRestartAttempt(instance)
 394           logging.error("Could not restart %s after %d attempts, giving up",
 395                         instance.name, MAXTRIES)
 396           continue
 397         try:
 398           logging.info("Restarting %s%s",
 399                         instance.name, last)
 400           instance.Restart()
 401           self.started_instances.add(instance.name)
 402         except Error, err:
 403           logging.error(str(err), exc_info=True)
 404
 405         notepad.RecordRestartAttempt(instance)
 406       elif instance.state in HELPLESS_STATES:
 407         if notepad.NumberOfRestartAttempts(instance):
 408           notepad.RemoveInstance(instance)
 409       else:
 410         if notepad.NumberOfRestartAttempts(instance):
 411           notepad.RemoveInstance(instance)
 412           logging.info("Restart of %s succeeded", instance.name)
 413
 414   def VerifyDisks(self):
 415     """Run gnt-cluster verify-disks.
 416
 417     """
 418     result = DoCmd(['gnt-cluster', 'verify-disks', '--lock-retries=15'])
 419     if result.output:
 420       logging.info(result.output)
 421
 422
 423 def ParseOptions():
 424   """Parse the command line options.
 425
 426   Returns:
 427     (options, args) as from OptionParser.parse_args()
 428
 429   """
 430   parser = OptionParser(description="Ganeti cluster watcher",
 431                         usage="%prog [-d]",
 432                         version="%%prog (ganeti) %s" %
 433                         constants.RELEASE_VERSION)
 434
 435   parser.add_option("-d", "--debug", dest="debug",
 436                     help="Write all messages to stderr",
 437                     default=False, action="store_true")
 438   options, args = parser.parse_args()
 439   return options, args
 440
 441
 442 def SetupLogging(debug):
 443   """Configures the logging module.
 444
 445   """
 446   formatter = logging.Formatter("%(asctime)s: %(message)s")
 447
 448   logfile_handler = logging.FileHandler(constants.LOG_WATCHER)
 449   logfile_handler.setFormatter(formatter)
 450   logfile_handler.setLevel(logging.INFO)
 451
 452   stderr_handler = logging.StreamHandler()
 453   stderr_handler.setFormatter(formatter)
 454   if debug:
 455     stderr_handler.setLevel(logging.NOTSET)
 456   else:
 457     stderr_handler.setLevel(logging.CRITICAL)
 458
 459   root_logger = logging.getLogger("")
 460   root_logger.setLevel(logging.NOTSET)
 461   root_logger.addHandler(logfile_handler)
 462   root_logger.addHandler(stderr_handler)
 463
 464
 465 def main():
 466   """Main function.
 467
 468   """
 469   options, args = ParseOptions()
 470
 471   SetupLogging(options.debug)
 472
 473   try:
 474     try:
 475       watcher = Watcher()
 476     except errors.ConfigurationError:
 477       # Just exit if there's no configuration
 478       sys.exit(constants.EXIT_SUCCESS)
 479     watcher.Run()
 480   except SystemExit:
 481     raise
 482   except NotMasterError:
 483     logging.debug("Not master, exiting")
 484     sys.exit(constants.EXIT_NOTMASTER)
 485   except errors.ResolverError, err:
 486     logging.error("Cannot resolve hostname '%s', exiting.", err.args[0])
 487     sys.exit(constants.EXIT_NODESETUP_ERROR)
 488   except Exception, err:
 489     logging.error(str(err), exc_info=True)
 490     sys.exit(constants.EXIT_FAILURE)
 491
 492
 493 if __name__ == '__main__':
 494   main()