#
#

# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


"""Tool to restart erroneously downed virtual machines.

This program and set of classes implement a watchdog to restart
virtual machines in a Ganeti cluster that have crashed or been killed
by a node reboot.  Run from cron or similar.

"""

import os
import os.path
import sys
import signal
import time
import logging
import errno
from optparse import OptionParser

from ganeti import utils
from ganeti import wconfd
from ganeti import constants
from ganeti import compat
from ganeti import errors
from ganeti import opcodes
from ganeti import cli
import ganeti.rpc.errors as rpcerr
from ganeti import rapi
from ganeti import netutils
from ganeti import qlang
from ganeti import ssconf
from ganeti import ht
from ganeti import pathutils

import ganeti.rapi.client # pylint: disable=W0611
from ganeti.rapi.client import UsesRapiClient

from ganeti.watcher import nodemaint
from ganeti.watcher import state


MAXTRIES = 5
BAD_STATES = compat.UniqueFrozenset([
  constants.INSTST_ERRORDOWN,
  ])
HELPLESS_STATES = compat.UniqueFrozenset([
  constants.INSTST_NODEDOWN,
  constants.INSTST_NODEOFFLINE,
  ])
NOTICE = "NOTICE"
ERROR = "ERROR"

#: Number of seconds to wait between starting child processes for node groups
CHILD_PROCESS_DELAY = 1.0

#: How many seconds to wait for instance status file lock
INSTANCE_STATUS_LOCK_TIMEOUT = 10.0


class NotMasterError(errors.GenericError):
  """Exception raised when this host is not the master."""


def ShouldPause():
  """Check whether we should pause.

  """
  return bool(utils.ReadWatcherPauseFile(pathutils.WATCHER_PAUSEFILE))


def StartNodeDaemons():
  """Start all the daemons that should be running on all nodes.

  """
  # on master or not, try to start the node daemon
  utils.EnsureDaemon(constants.NODED)
  # start confd as well. On non candidates it will be in disabled mode.
  utils.EnsureDaemon(constants.CONFD)
  # start mond as well: all nodes need monitoring
  if constants.ENABLE_MOND:
    utils.EnsureDaemon(constants.MOND)
  # start kvmd, which will quit if not needed to run
  utils.EnsureDaemon(constants.KVMD)


def RunWatcherHooks():
  """Run the watcher hooks.

  """
  hooks_dir = utils.PathJoin(pathutils.HOOKS_BASE_DIR,
                             constants.HOOKS_NAME_WATCHER)
  if not os.path.isdir(hooks_dir):
    return

  try:
    results = utils.RunParts(hooks_dir)
  except Exception, err: # pylint: disable=W0703
    logging.exception("RunParts %s failed: %s", hooks_dir, err)
    return

  for (relname, status, runresult) in results:
    if status == constants.RUNPARTS_SKIP:
      logging.debug("Watcher hook %s: skipped", relname)
    elif status == constants.RUNPARTS_ERR:
      logging.warning("Watcher hook %s: error (%s)", relname, runresult)
    elif status == constants.RUNPARTS_RUN:
      if runresult.failed:
        logging.warning("Watcher hook %s: failed (exit: %d) (output: %s)",
                        relname, runresult.exit_code, runresult.output)
      else:
        logging.debug("Watcher hook %s: success (output: %s)", relname,
                      runresult.output)
    else:
      raise errors.ProgrammerError("Unknown status %s returned by RunParts",
                                   status)


class Instance(object):
  """Abstraction for a Virtual Machine instance.

  """
  def __init__(self, name, status, config_state, config_state_source,
               disks_active, snodes, disk_template):
    self.name = name
    self.status = status
    self.config_state = config_state
    self.config_state_source = config_state_source
    self.disks_active = disks_active
    self.snodes = snodes
    self.disk_template = disk_template

  def Restart(self, cl):
    """Encapsulates the start of an instance.

    """
    op = opcodes.OpInstanceStartup(instance_name=self.name, force=False)
    op.reason = [(constants.OPCODE_REASON_SRC_WATCHER,
                  "Restarting instance %s" % self.name,
                  utils.EpochNano())]
    cli.SubmitOpCode(op, cl=cl)

  def ActivateDisks(self, cl):
    """Encapsulates the activation of all disks of an instance.

    """
    op = opcodes.OpInstanceActivateDisks(instance_name=self.name)
    op.reason = [(constants.OPCODE_REASON_SRC_WATCHER,
                  "Activating disks for instance %s" % self.name,
                  utils.EpochNano())]
    cli.SubmitOpCode(op, cl=cl)

  def NeedsCleanup(self):
    """Determines whether the instance needs cleanup.

    Determines whether the instance needs cleanup after having been
    shutdown by the user.

    @rtype: bool
    @return: True if the instance needs cleanup, False otherwise.

    """
    return self.status == constants.INSTST_USERDOWN and \
        self.config_state != constants.ADMINST_DOWN


class Node(object):
  """Data container representing cluster node.

  """
  def __init__(self, name, bootid, offline, secondaries):
    """Initializes this class.

    """
    self.name = name
    self.bootid = bootid
    self.offline = offline
    self.secondaries = secondaries


def _CleanupInstance(cl, notepad, inst, locks):
  n = notepad.NumberOfCleanupAttempts(inst.name)

  if inst.name in locks:
    logging.info("Not cleaning up instance '%s', instance is locked",
                 inst.name)
    return

  if n > MAXTRIES:
    logging.warning("Not cleaning up instance '%s', retries exhausted",
                    inst.name)
    return

  logging.info("Instance '%s' was shutdown by the user, cleaning up instance",
               inst.name)
  op = opcodes.OpInstanceShutdown(instance_name=inst.name,
                                  admin_state_source=constants.USER_SOURCE)

  op.reason = [(constants.OPCODE_REASON_SRC_WATCHER,
                "Cleaning up instance %s" % inst.name,
                utils.EpochNano())]
  try:
    cli.SubmitOpCode(op, cl=cl)
    if notepad.NumberOfCleanupAttempts(inst.name):
      notepad.RemoveInstance(inst.name)
  except Exception: # pylint: disable=W0703
    logging.exception("Error while cleaning up instance '%s'", inst.name)
    notepad.RecordCleanupAttempt(inst.name)


def _CheckInstances(cl, notepad, instances, locks):
  """Make a pass over the list of instances, restarting downed ones.

  """
  notepad.MaintainInstanceList(instances.keys())

  started = set()

  for inst in instances.values():
    if inst.NeedsCleanup():
      _CleanupInstance(cl, notepad, inst, locks)
    elif inst.status in BAD_STATES:
      n = notepad.NumberOfRestartAttempts(inst.name)

      if n > MAXTRIES:
        logging.warning("Not restarting instance '%s', retries exhausted",
                        inst.name)
        continue

      if n == MAXTRIES:
        notepad.RecordRestartAttempt(inst.name)
        logging.error("Could not restart instance '%s' after %s attempts,"
                      " giving up", inst.name, MAXTRIES)
        continue

      try:
        logging.info("Restarting instance '%s' (attempt #%s)",
                     inst.name, n + 1)
        inst.Restart(cl)
      except Exception: # pylint: disable=W0703
        logging.exception("Error while restarting instance '%s'", inst.name)
      else:
        started.add(inst.name)

      notepad.RecordRestartAttempt(inst.name)

    else:
      if notepad.NumberOfRestartAttempts(inst.name):
        notepad.RemoveInstance(inst.name)
        if inst.status not in HELPLESS_STATES:
          logging.info("Restart of instance '%s' succeeded", inst.name)

  return started


def _CheckDisks(cl, notepad, nodes, instances, started):
  """Check all nodes for restarted ones.

  """
  check_nodes = []

  for node in nodes.values():
    old = notepad.GetNodeBootID(node.name)
    if not node.bootid:
      # Bad node, not returning a boot id
      if not node.offline:
        logging.debug("Node '%s' missing boot ID, skipping secondary checks",
                      node.name)
      continue

    if old != node.bootid:
      # Node's boot ID has changed, probably through a reboot
      check_nodes.append(node)

  if check_nodes:
    # Activate disks for all instances with any of the checked nodes as a
    # secondary node.
    for node in check_nodes:
      for instance_name in node.secondaries:
        try:
          inst = instances[instance_name]
        except KeyError:
          logging.info("Can't find instance '%s', maybe it was ignored",
                       instance_name)
          continue

        if not inst.disks_active:
          logging.info("Skipping disk activation for instance with not"
                       " activated disks '%s'", inst.name)
          continue

        if inst.name in started:
          # we already tried to start the instance, which should have
          # activated its drives (if they can be at all)
          logging.debug("Skipping disk activation for instance '%s' as"
                        " it was already started", inst.name)
          continue

        try:
          logging.info("Activating disks for instance '%s'", inst.name)
          inst.ActivateDisks(cl)
        except Exception: # pylint: disable=W0703
          logging.exception("Error while activating disks for instance '%s'",
                            inst.name)

    # Keep changed boot IDs
    for node in check_nodes:
      notepad.SetNodeBootID(node.name, node.bootid)


def _CheckForOfflineNodes(nodes, instance):
  """Checks if given instances has any secondary in offline status.

  @param instance: The instance object
  @return: True if any of the secondary is offline, False otherwise

  """
  return compat.any(nodes[node_name].offline for node_name in instance.snodes)


def _GetPendingVerifyDisks(cl, uuid):
  """Checks if there are any currently running or pending group verify jobs and
  if so, returns their id.

  """
  qfilter = qlang.MakeSimpleFilter("status",
                                    frozenset([constants.JOB_STATUS_RUNNING,
                                               constants.JOB_STATUS_QUEUED,
                                               constants.JOB_STATUS_WAITING]))
  qresult = cl.Query(constants.QR_JOB, ["id", "summary"], qfilter)

  ids = [jobid for ((_, jobid), (_, (job, ))) in qresult.data
         if job == ("GROUP_VERIFY_DISKS(%s)" % uuid)]
  return ids


def _VerifyDisks(cl, uuid, nodes, instances, is_strict):
  """Run a per-group "gnt-cluster verify-disks".

  """

  existing_jobs = _GetPendingVerifyDisks(cl, uuid)
  if existing_jobs:
    logging.info("There are verify disks jobs already pending (%s), skipping "
                 "VerifyDisks step for %s.",
                 utils.CommaJoin(existing_jobs), uuid)
    return

  op = opcodes.OpGroupVerifyDisks(
    group_name=uuid, priority=constants.OP_PRIO_LOW, is_strict=is_strict)
  op.reason = [(constants.OPCODE_REASON_SRC_WATCHER,
                "Verifying disks of group %s" % uuid,
                utils.EpochNano())]
  job_id = cl.SubmitJob([op])
  ((_, offline_disk_instances, _), ) = \
    cli.PollJob(job_id, cl=cl, feedback_fn=logging.debug)
  cl.ArchiveJob(job_id)

  if not offline_disk_instances:
    # nothing to do
    logging.debug("Verify-disks reported no offline disks, nothing to do")
    return

  logging.debug("Will activate disks for instance(s) %s",
                utils.CommaJoin(offline_disk_instances))

  # We submit only one job, and wait for it. Not optimal, but this puts less
  # load on the job queue.
  job = []
  for name in offline_disk_instances:
    try:
      inst = instances[name]
    except KeyError:
      logging.info("Can't find instance '%s', maybe it was ignored", name)
      continue

    if inst.status in HELPLESS_STATES or _CheckForOfflineNodes(nodes, inst):
      logging.info("Skipping instance '%s' because it is in a helpless state"
                   " or has offline secondaries", name)
      continue

    op = opcodes.OpInstanceActivateDisks(instance_name=name)
    op.reason = [(constants.OPCODE_REASON_SRC_WATCHER,
                  "Activating disks for instance %s" % name,
                  utils.EpochNano())]
    job.append(op)

  if job:
    job_id = cli.SendJob(job, cl=cl)

    try:
      cli.PollJob(job_id, cl=cl, feedback_fn=logging.debug)
    except Exception: # pylint: disable=W0703
      logging.exception("Error while activating disks")


def IsRapiResponding(hostname):
  """Connects to RAPI port and does a simple test.

  Connects to RAPI port of hostname and does a simple test. At this time, the
  test is GetVersion.

  If RAPI responds with error code "401 Unauthorized", the test is successful,
  because the aim of this function is to assess whether RAPI is responding, not
  if it is accessible.

  @type hostname: string
  @param hostname: hostname of the node to connect to.
  @rtype: bool
  @return: Whether RAPI is working properly

  """
  curl_config = rapi.client.GenericCurlConfig()
  rapi_client = rapi.client.GanetiRapiClient(hostname,
                                             curl_config_fn=curl_config)
  try:
    master_version = rapi_client.GetVersion()
  except rapi.client.CertificateError, err:
    logging.warning("RAPI certificate error: %s", err)
    return False
  except rapi.client.GanetiApiError, err:
    if err.code == 401:
      # Unauthorized, but RAPI is alive and responding
      return True
    else:
      logging.warning("RAPI error: %s", err)
      return False
  else:
    logging.debug("Reported RAPI version %s", master_version)
    return master_version == constants.RAPI_VERSION


def IsWconfdResponding():
  """Probes an echo RPC to WConfD.

  """
  probe_string = "ganeti watcher probe %d" % time.time()

  try:
    result = wconfd.Client().Echo(probe_string)
  except Exception, err: # pylint: disable=W0703
    logging.warning("WConfd connection error: %s", err)
    return False

  if result != probe_string:
    logging.warning("WConfd echo('%s') returned '%s'", probe_string, result)
    return False

  return True


def ParseOptions():
  """Parse the command line options.

  @return: (options, args) as from OptionParser.parse_args()

  """
  parser = OptionParser(description="Ganeti cluster watcher",
                        usage="%prog [-d]",
                        version="%%prog (ganeti) %s" %
                        constants.RELEASE_VERSION)

  parser.add_option(cli.DEBUG_OPT)
  parser.add_option(cli.NODEGROUP_OPT)
  parser.add_option("-A", "--job-age", dest="job_age", default=6 * 3600,
                    help="Autoarchive jobs older than this age (default"
                          " 6 hours)")
  parser.add_option("--ignore-pause", dest="ignore_pause", default=False,
                    action="store_true", help="Ignore cluster pause setting")
  parser.add_option("--wait-children", dest="wait_children",
                    action="store_true", help="Wait for child processes")
  parser.add_option("--no-wait-children", dest="wait_children",
                    action="store_false",
                    help="Don't wait for child processes")
  parser.add_option("--no-verify-disks", dest="no_verify_disks", default=False,
                    action="store_true", help="Do not verify disk status")
  parser.add_option("--no-strict", dest="no_strict",
                    default=False, action="store_true",
                    help="Do not run group verify in strict mode")
  parser.add_option("--rapi-ip", dest="rapi_ip",
                    default=constants.IP4_ADDRESS_LOCALHOST,
                    help="Use this IP to talk to RAPI.")
  # See optparse documentation for why default values are not set by options
  parser.set_defaults(wait_children=True)
  options, args = parser.parse_args()
  options.job_age = cli.ParseTimespec(options.job_age)

  if args:
    parser.error("No arguments expected")

  return (options, args)


def _WriteInstanceStatus(filename, data):
  """Writes the per-group instance status file.

  The entries are sorted.

  @type filename: string
  @param filename: Path to instance status file
  @type data: list of tuple; (instance name as string, status as string)
  @param data: Instance name and status

  """
  logging.debug("Updating instance status file '%s' with %s instances",
                filename, len(data))

  utils.WriteFile(filename,
                  data="\n".join("%s %s" % (n, s) for (n, s) in sorted(data)))


def _UpdateInstanceStatus(filename, instances):
  """Writes an instance status file from L{Instance} objects.

  @type filename: string
  @param filename: Path to status file
  @type instances: list of L{Instance}

  """
  _WriteInstanceStatus(filename, [(inst.name, inst.status)
                                  for inst in instances])


def _ReadInstanceStatus(filename):
  """Reads an instance status file.

  @type filename: string
  @param filename: Path to status file
  @rtype: tuple; (None or number, list of lists containing instance name and
    status)
  @return: File's mtime and instance status contained in the file; mtime is
    C{None} if file can't be read

  """
  logging.debug("Reading per-group instance status from '%s'", filename)

  statcb = utils.FileStatHelper()
  try:
    content = utils.ReadFile(filename, preread=statcb)
  except EnvironmentError, err:
    if err.errno == errno.ENOENT:
      logging.error("Can't read '%s', does not exist (yet)", filename)
    else:
      logging.exception("Unable to read '%s', ignoring", filename)
    return (None, None)
  else:
    return (statcb.st.st_mtime, [line.split(None, 1)
                                 for line in content.splitlines()])


def _MergeInstanceStatus(filename, pergroup_filename, groups):
  """Merges all per-group instance status files into a global one.

  @type filename: string
  @param filename: Path to global instance status file
  @type pergroup_filename: string
  @param pergroup_filename: Path to per-group status files, must contain "%s"
    to be replaced with group UUID
  @type groups: sequence
  @param groups: UUIDs of known groups

  """
  # Lock global status file in exclusive mode
  lock = utils.FileLock.Open(filename)
  try:
    lock.Exclusive(blocking=True, timeout=INSTANCE_STATUS_LOCK_TIMEOUT)
  except errors.LockError, err:
    # All per-group processes will lock and update the file. None of them
    # should take longer than 10 seconds (the value of
    # INSTANCE_STATUS_LOCK_TIMEOUT).
    logging.error("Can't acquire lock on instance status file '%s', not"
                  " updating: %s", filename, err)
    return

  logging.debug("Acquired exclusive lock on '%s'", filename)

  data = {}

  # Load instance status from all groups
  for group_uuid in groups:
    (mtime, instdata) = _ReadInstanceStatus(pergroup_filename % group_uuid)

    if mtime is not None:
      for (instance_name, status) in instdata:
        data.setdefault(instance_name, []).append((mtime, status))

  # Select last update based on file mtime
  inststatus = [(instance_name, sorted(status, reverse=True)[0][1])
                for (instance_name, status) in data.items()]

  # Write the global status file. Don't touch file after it's been
  # updated--there is no lock anymore.
  _WriteInstanceStatus(filename, inststatus)


def GetLuxiClient(try_restart):
  """Tries to connect to the luxi daemon.

  @type try_restart: bool
  @param try_restart: Whether to attempt to restart the master daemon

  """
  try:
    return cli.GetClient()
  except errors.OpPrereqError, err:
    # this is, from cli.GetClient, a not-master case
    raise NotMasterError("Not on master node (%s)" % err)

  except (rpcerr.NoMasterError, rpcerr.TimeoutError), err:
    if not try_restart:
      raise

    logging.warning("Luxi daemon seems to be down (%s), trying to restart",
                    err)

    if not utils.EnsureDaemon(constants.LUXID):
      raise errors.GenericError("Can't start the master daemon")

    # Retry the connection
    return cli.GetClient()


def _StartGroupChildren(cl, wait):