src/kernel/angel.cpp

/* Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License, version 2.0,
   as published by the Free Software Foundation.

   This program is also distributed with certain software (including
   but not limited to OpenSSL) that is licensed under separate terms,
   as designated in a particular file or component or in included license
   documentation.  The authors of MySQL hereby grant you an additional
   permission to link the program and your derivative works with the
   separately licensed software that they have included with MySQL.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License, version 2.0, for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA */


#include <ndb_global.h>
#include <ndb_version.h>

#include "angel.hpp"
#include "ndbd.hpp"

#include <NdbConfig.h>
#include <NdbAutoPtr.hpp>
#include <portlib/ndb_daemon.h>
#include <portlib/NdbSleep.h>
#include <portlib/NdbDir.hpp>

#include <ConfigRetriever.hpp>

#include <EventLogger.hpp>
extern EventLogger * g_eventLogger;

static void
angel_exit(int code)
{
  ndb_daemon_exit(code);
}

#include "../mgmapi/mgmapi_configuration.hpp"

static void
reportShutdown(const ndb_mgm_configuration* config,
               NodeId nodeid, int error_exit,
               bool restart, bool nostart, bool initial,
               Uint32 error, Uint32 signum, Uint32 sphase)
{
  // Only allow "initial" and "nostart" to be set if "restart" is set
  assert(restart ||
         (!restart && !initial && !nostart));

  Uint32 length, theData[25];
  EventReport *rep= CAST_PTR(EventReport, &theData[0]);

  rep->setNodeId(nodeid);
  if (restart)
    theData[1]=1 |
      (nostart ? 2 : 0) |
      (initial ? 4 : 0);
  else
    theData[1]=0;

  if (error_exit == 0)
  {
    rep->setEventType(NDB_LE_NDBStopCompleted);
    theData[2]=signum;
    length=3;
  } else
  {
    rep->setEventType(NDB_LE_NDBStopForced);
    theData[2]=signum;
    theData[3]=error;
    theData[4]=sphase;
    theData[5]=0; // extra
    length=6;
  }

  // Log event locally
  g_eventLogger->log(rep->getEventType(), theData, length,
                     rep->getNodeId(), 0);

  // Log event to cluster log
  ndb_mgm_configuration_iterator iter(*config, CFG_SECTION_NODE);
  for (iter.first(); iter.valid(); iter.next())
  {
    Uint32 type;
    if (iter.get(CFG_TYPE_OF_SECTION, &type) ||
       type != NODE_TYPE_MGM)
      continue;

    Uint32 port;
    if (iter.get(CFG_MGM_PORT, &port))
      continue;

    const char* hostname;
    if (iter.get(CFG_NODE_HOST, &hostname))
      continue;

    BaseString connect_str;
    connect_str.assfmt("%s:%d", hostname, port);


    NdbMgmHandle h = ndb_mgm_create_handle();
    if (h == 0)
    {
      g_eventLogger->warning("Unable to report shutdown reason "
                             "to '%s'(failed to create mgm handle)",
                             connect_str.c_str());
      continue;
    }

    if (ndb_mgm_set_connectstring(h, connect_str.c_str()) ||
        ndb_mgm_connect(h, 1, 0, 0) ||
        ndb_mgm_report_event(h, theData, length))
    {
      g_eventLogger->warning("Unable to report shutdown reason "
                             "to '%s'(error: %s - %s)",
                             connect_str.c_str(),
                             ndb_mgm_get_latest_error_msg(h),
                             ndb_mgm_get_latest_error_desc(h));
    }

    ndb_mgm_destroy_handle(&h);
  }
}


static void
ignore_signals(void)
{
  static const int ignore_list[] = {
#ifdef SIGBREAK
    SIGBREAK,
#endif
#ifdef SIGHUP
    SIGHUP,
#endif
    SIGINT,
#if defined SIGPWR
    SIGPWR,
#elif defined SIGINFO
    SIGINFO,
#endif
    SIGQUIT,
    SIGTERM,
#ifdef SIGTSTP
    SIGTSTP,
#endif
#ifdef SIGTTIN
    SIGTTIN,
#endif
#ifdef SIGTTOU
    SIGTTOU,
#endif
    SIGABRT,
#ifdef SIGALRM
    SIGALRM,
#endif
#ifdef SIGBUS
    SIGBUS,
#endif
    SIGFPE,
    SIGILL,
#ifdef SIGIO
    SIGIO,
#endif
#ifdef SIGPOLL
    SIGPOLL,
#endif
    SIGSEGV,
    SIGPIPE,
#ifdef SIGTRAP
    SIGTRAP
#endif
  };

  for(size_t i = 0; i < sizeof(ignore_list)/sizeof(ignore_list[0]); i++)
    signal(ignore_list[i], SIG_IGN);
}

#ifdef _WIN32
static inline
int pipe(int pipefd[2]){
  const unsigned int buffer_size = 4096;
  const int flags = 0;
  return _pipe(pipefd, buffer_size, flags);
}

#undef getpid
#include <process.h>

typedef DWORD pid_t;

static const int WNOHANG = 37;

static inline
pid_t waitpid(pid_t pid, int *stat_loc, int options)
{
  /* Only support waitpid(,,WNOHANG) */
  assert(options == WNOHANG);
  assert(stat_loc);

  HANDLE handle = OpenProcess(PROCESS_ALL_ACCESS, FALSE, pid);
  if (handle == NULL)
  {
    g_eventLogger->error("waitpid: Could not open handle for pid %d, "
                         "error: %d", pid, GetLastError());
    return -1;
  }

  DWORD exit_code;
  if (!GetExitCodeProcess(handle, &exit_code))
  {
    g_eventLogger->error("waitpid: GetExitCodeProcess failed, pid: %d, "
                         "error: %d", pid, GetLastError());
    CloseHandle(handle);
    return -1;
  }
  CloseHandle(handle);

  if (exit_code == STILL_ACTIVE)
  {
    /* Still alive */
    return 0;
  }

  *stat_loc = exit_code;

  return pid;
}

static inline
bool WIFEXITED(int status)
{
  return true;
}

static inline
int WEXITSTATUS(int status)
{
  return status;
}

static inline
bool WIFSIGNALED(int status)
{
  return false;
}

static inline
int WTERMSIG(int status)
{
  return 0;
}

static int
kill(pid_t pid, int sig)
{
  int retry_open_event = 10;

  char shutdown_event_name[32];
  _snprintf(shutdown_event_name, sizeof(shutdown_event_name),
            "ndbd_shutdown_%d", pid);

  /* Open the event to signal */
  HANDLE shutdown_event;
  while ((shutdown_event =
          OpenEvent(EVENT_MODIFY_STATE, FALSE, shutdown_event_name)) == NULL)
  {
     /*
      Check if the process is alive, otherwise there is really
      no sense to retry the open of the event
     */
    DWORD exit_code;
    HANDLE process = OpenProcess(SYNCHRONIZE | PROCESS_QUERY_INFORMATION,
                                  FALSE, pid);
    if (!process)
    {
      /* Already died */
      return -1;
    }

    if (!GetExitCodeProcess(process,&exit_code))
    {
      g_eventLogger->error("GetExitCodeProcess failed, pid: %d, error: %d",
                           pid, GetLastError());
      CloseHandle(process);
      return -1;
    }
    CloseHandle(process);

    if (exit_code != STILL_ACTIVE)
    {
      /* Already died */
      return -1;
    }

    if (retry_open_event--)
      Sleep(100);
    else
    {
      g_eventLogger->error("Failed to open shutdown_event '%s', error: %d",
                            shutdown_event_name, GetLastError());
      return -1;
    }
  }

  if (SetEvent(shutdown_event) == 0)
  {
    g_eventLogger->error("Failed to signal shutdown_event '%s', error: %d",
                         shutdown_event_name, GetLastError());
  }
  CloseHandle(shutdown_event);
  return pid;
}
#endif

extern int real_main(int, char**);


static
char** create_argv(const Vector<BaseString>& args)
{
  char **argv = (char **)malloc(sizeof(char*) * (args.size() + 1));
  if(argv == NULL)
    return NULL;

  for(unsigned i = 0; i < args.size(); i++)
    argv[i] = strdup(args[i].c_str());
  argv[args.size()] = NULL;
  return argv;
}


static
void free_argv(char** argv)
{
  char** argp = argv;
  while(*argp)
  {
    free((void*)*argp);
    argp++;
  }
  free((void*)argv);
}


static pid_t
spawn_process(const char* progname, const Vector<BaseString>& args)
{
#ifdef _WIN32
  // Get full path name of this executeble
  char path[MAX_PATH];
  DWORD len = GetModuleFileName(NULL, path, sizeof(path));
  if (len == 0 || len == sizeof(path))
  {
    g_eventLogger->warning("spawn_process: Could not extract full path, "
                           "len: %u, error: %u\n",
                           len, GetLastError());
    // Fall through and try with progname as it was supplied
  }
  else
  {
    progname = path;
  }
#endif

  char** argv = create_argv(args);
  if (!argv)
  {
    g_eventLogger->error("spawn_process: Failed to create argv, errno: %d",
                         errno);
    return -1;
  }

#ifdef _WIN32

  intptr_t spawn_handle = _spawnv(P_NOWAIT, progname, argv);
  if (spawn_handle == -1)
  {
    g_eventLogger->error("spawn_process: Failed to spawn process, errno: %d",
                         errno);
    // Print the _spawnv arguments to aid debugging
    g_eventLogger->error(" progname: '%s'", progname);
    char** argp = argv;
    while(*argp)
      g_eventLogger->error("argv: '%s'", *argp++);

    free_argv(argv);
    return -1;
  }
  free_argv(argv);

  // Convert the handle returned from spawnv_ to a pid
  DWORD pid = GetProcessId((HANDLE)spawn_handle);
  if (pid == 0)
  {
    g_eventLogger->error("spawn_process: Failed to convert handle %d "
                         "to pid, error: %d", spawn_handle, GetLastError());
    CloseHandle((HANDLE)spawn_handle);
    return -1;
  }
  CloseHandle((HANDLE)spawn_handle);
  return pid;
#else
  pid_t pid = fork();
  if (pid == -1)
  {
    g_eventLogger->error("Failed to fork, errno: %d", errno);
    free_argv(argv);
    return -1;
  }

  if (pid)
  {
    free_argv(argv);
    // Parent
    return pid;
  }

  // Count number of arguments
  int argc = 0;
  while(argv[argc])
    argc++;

  // Calling 'main' to start program from beginning
  // without loading (possibly new version) from disk
  (void)real_main(argc, argv);
  assert(false); // main should never return
  exit(1);
  return -1; // Never reached
#endif
}

/*
  retry failed spawn after sleep until fork suceeds or
  max number of retries occurs
*/

static pid_t
retry_spawn_process(const char* progname, const Vector<BaseString>& args)
{
  const unsigned max_retries = 10;
  unsigned retry_counter = 0;
  while(true)
  {
    pid_t pid = spawn_process(progname, args);
    if (pid == -1)
    {
      if (retry_counter++ == max_retries)
      {
        g_eventLogger->error("Angel failed to spawn %d times, giving up",
                             retry_counter);
        angel_exit(1);
      }

      g_eventLogger->warning("Angel failed to spawn, sleep and retry");

      NdbSleep_SecSleep(1);
      continue;
    }
    return pid;
  }
}

static Uint32 stop_on_error;
static Uint32 config_max_start_fail_retries;
static Uint32 config_restart_delay_secs;


/*
  Extract the config parameters that concerns angel
*/

static bool
configure(const ndb_mgm_configuration* conf, NodeId nodeid)
{
  Uint32 generation = 0;
  ndb_mgm_configuration_iterator sys_iter(*conf, CFG_SECTION_SYSTEM);
  if (sys_iter.get(CFG_SYS_CONFIG_GENERATION, &generation))
  {
    g_eventLogger->warning("Configuration didn't contain generation "
                           "(likely old ndb_mgmd");
  }
  g_eventLogger->debug("Using configuration with generation %u", generation);

  ndb_mgm_configuration_iterator iter(*conf, CFG_SECTION_NODE);
  if (iter.find(CFG_NODE_ID, nodeid))
  {
    g_eventLogger->error("Invalid configuration fetched, could not "
                         "find own node id %d", nodeid);
    return false;
  }

  if (iter.get(CFG_DB_STOP_ON_ERROR, &stop_on_error))
  {
    g_eventLogger->error("Invalid configuration fetched, could not "
                         "find StopOnError");
    return false;
  }
  g_eventLogger->debug("Using StopOnError: %u", stop_on_error);

  if (iter.get(CFG_DB_MAX_START_FAIL, &config_max_start_fail_retries))
  {
    /* Old Management node, use default value */
    config_max_start_fail_retries = 3;
  }

  if (iter.get(CFG_DB_START_FAIL_DELAY_SECS, &config_restart_delay_secs))
  {
    /* Old Management node, use default value */
    config_restart_delay_secs = 0;
  }

  const char * datadir;
  if (iter.get(CFG_NODE_DATADIR, &datadir))
  {
    g_eventLogger->error("Invalid configuration fetched, could not "
                         "find DataDir");
    return false;
  }
  g_eventLogger->debug("Using DataDir: %s", datadir);

  NdbConfig_SetPath(datadir);

  if (NdbDir::chdir(NdbConfig_get_path(NULL)) != 0)
  {
    g_eventLogger->warning("Cannot change directory to '%s', error: %d",
                           NdbConfig_get_path(NULL), errno);
    // Ignore error
  }

  return true;
}

bool stop_child = false;

void
angel_run(const char* progname,
          const Vector<BaseString>& original_args,
          const char* connect_str,
          int force_nodeid,
          const char* bind_address,
          bool initial,
          bool no_start,
          bool daemon)
{
  ConfigRetriever retriever(connect_str,
                            force_nodeid,
                            NDB_VERSION,
                            NDB_MGM_NODE_TYPE_NDB,
                            bind_address);
  if (retriever.hasError())
  {
    g_eventLogger->error("Could not initialize connection to management "
                         "server, error: '%s'", retriever.getErrorString());
    angel_exit(1);
  }

  const int connnect_retries = 12;
  const int connect_delay = 5;
  const int verbose = 1;
  if (retriever.do_connect(connnect_retries, connect_delay, verbose) != 0)
  {
    g_eventLogger->error("Could not connect to management server, "
                         "error: '%s'", retriever.getErrorString());
    angel_exit(1);
  }
  g_eventLogger->info("Angel connected to '%s:%d'",
                      retriever.get_mgmd_host(),
                      retriever.get_mgmd_port());

  const int alloc_retries = 2;
  const int alloc_delay = 3;
  const Uint32 nodeid = retriever.allocNodeId(alloc_retries, alloc_delay);
  if (nodeid == 0)
  {
    g_eventLogger->error("Failed to allocate nodeid, error: '%s'",
                         retriever.getErrorString());
    angel_exit(1);
  }
  g_eventLogger->info("Angel allocated nodeid: %u", nodeid);

  ndb_mgm_configuration * config = retriever.getConfig(nodeid);
  NdbAutoPtr<ndb_mgm_configuration> config_autoptr(config);
  if (config == 0)
  {
    g_eventLogger->error("Could not fetch configuration/invalid "
                         "configuration, error: '%s'",
                         retriever.getErrorString());
    angel_exit(1);
  }

  if (!configure(config, nodeid))
  {
    // Failed to configure, error already printed
    angel_exit(1);
  }

  if (daemon)
  {
    // Become a daemon
    char *lockfile = NdbConfig_PidFileName(nodeid);
    char *logfile = NdbConfig_StdoutFileName(nodeid);
    NdbAutoPtr<char> tmp_aptr1(lockfile), tmp_aptr2(logfile);

    if (ndb_daemonize(lockfile, logfile) != 0)
    {
      g_eventLogger->error("Couldn't start as daemon, error: '%s'",
                           ndb_daemon_error);
      angel_exit(1);
    }
  }

  // Counter for consecutive failed startups
  Uint32 failed_startups_counter = 0;
  while (true)
  {

    // Create pipe where ndbd process will report extra shutdown status
    int fds[2];
    if (pipe(fds))
    {
      g_eventLogger->error("Failed to create pipe, errno: %d (%s)",
                           errno, strerror(errno));
      angel_exit(1);
    }

    FILE *child_info_r;
    if (!(child_info_r = fdopen(fds[0], "r")))
    {
      g_eventLogger->error("Failed to open stream for pipe, errno: %d (%s)",
                           errno, strerror(errno));
      angel_exit(1);
    }

    // Build the args used to start ndbd by appending
    // the arguments that may have changed at the end
    // of original argument list
    BaseString one_arg;
    Vector<BaseString> args;
    args = original_args;

    // Pass fd number of the pipe which ndbd should use
    // for sending extra status to angel
    one_arg.assfmt("--report-fd=%d", fds[1]);
    args.push_back(one_arg);

    // The nodeid which has been allocated by angel
    one_arg.assfmt("--allocated-nodeid=%d", nodeid);
    args.push_back(one_arg);

    one_arg.assfmt("--initial=%d", initial);
    args.push_back(one_arg);

    one_arg.assfmt("--nostart=%d", no_start);
    args.push_back(one_arg);

    pid_t child = retry_spawn_process(progname, args);
    if (child <= 0)
    {
      // safety, retry_spawn_process returns valid child or give up
      g_eventLogger->error("retry_spawn_process, child: %d", child);
      angel_exit(1);
    }

    /**
     * Parent
     */
    g_eventLogger->info("Angel pid: %d started child: %d",
                        getpid(), child);

    ignore_signals();

    int status=0, error_exit=0;
    while(true)
    {
      pid_t ret_pid = waitpid(child, &status, WNOHANG);
      if (ret_pid == child)
      {
        g_eventLogger->debug("Angel got child %d", child);
        break;
      }
      if (ret_pid > 0)
      {
        g_eventLogger->warning("Angel got unexpected pid %d "
                               "when waiting for %d",
                               ret_pid, child);
      }

      if (stop_child)
      {
        g_eventLogger->info("Angel shutting down ndbd with pid %d", child);
        kill(child, SIGINT);
       }
      NdbSleep_MilliSleep(100);
    }

    // Close the write end of pipe
    close(fds[1]);

    // Read info from the child's pipe
    char buf[128];
    Uint32 child_error = 0, child_signal = 0, child_sphase = 0;
    while (fgets(buf, sizeof (buf), child_info_r))
    {
      int value;
      if (sscanf(buf, "error=%d\n", &value) == 1)
        child_error = value;
      else if (sscanf(buf, "signal=%d\n", &value) == 1)
        child_signal = value;
      else if (sscanf(buf, "sphase=%d\n", &value) == 1)
        child_sphase = value;
      else if (strcmp(buf, "\n") != 0)
        fprintf(stderr, "unknown info from child: '%s'\n", buf);
    }
    g_eventLogger->debug("error: %u, signal: %u, sphase: %u",
                         child_error, child_signal, child_sphase);
    // Close read end of pipe in parent
    fclose(child_info_r);

    if (WIFEXITED(status))
    {
      switch (WEXITSTATUS(status)) {
      case NRT_Default:
        g_eventLogger->info("Angel shutting down");
        reportShutdown(config, nodeid, 0, 0, false, false,
                       child_error, child_signal, child_sphase);
        angel_exit(0);
        break;
      case NRT_NoStart_Restart:
        initial = false;
        no_start = true;
        break;
      case NRT_NoStart_InitialStart:
        initial = true;
        no_start = true;
        break;
      case NRT_DoStart_InitialStart:
        initial = true;
        no_start = false;
        break;
      default:
        error_exit=1;
        if (stop_on_error)
        {
          /**
           * Error shutdown && stopOnError()
           */
          reportShutdown(config, nodeid,
                         error_exit, 0, false, false,
                         child_error, child_signal, child_sphase);
          angel_exit(0);
        }
        // Fall-through
      case NRT_DoStart_Restart:
        initial = false;
        no_start = false;
        break;
      }
    } else
    {
      error_exit=1;
      if (WIFSIGNALED(status))
      {
        child_signal = WTERMSIG(status);
      }
      else
      {
        child_signal = 127;
        g_eventLogger->info("Unknown exit reason. Stopped.");
      }
      if (stop_on_error)
      {
        /**
         * Error shutdown && stopOnError()
         */
        reportShutdown(config, nodeid,
                       error_exit, 0, false, false,
                       child_error, child_signal, child_sphase);
        angel_exit(0);
      }
    }

    // Check startup failure
    const Uint32 STARTUP_FAILURE_SPHASE = 6;
    Uint32 restart_delay_secs = 0;
    if (error_exit && // Only check startup failure if ndbd exited uncontrolled
        child_sphase <= STARTUP_FAILURE_SPHASE)
    {
      if (++failed_startups_counter >= config_max_start_fail_retries)
      {
        g_eventLogger->alert("Angel detected too many startup failures(%d), "
                             "not restarting again", failed_startups_counter);
        reportShutdown(config, nodeid,
                       error_exit, 0, false, false,
                       child_error, child_signal, child_sphase);
        angel_exit(0);
      }
      g_eventLogger->info("Angel detected startup failure, count: %u",
                          failed_startups_counter);

      restart_delay_secs = config_restart_delay_secs;
    }
    else
    {
      // Reset the counter for consecutive failed startups
      failed_startups_counter = 0;
    }

    reportShutdown(config, nodeid,
                   error_exit, 1,
                   no_start,
                   initial,
                   child_error, child_signal, child_sphase);
    g_eventLogger->info("Ndb has terminated (pid %d) restarting", child);

    g_eventLogger->debug("Angel reconnecting to management server");
    (void)retriever.disconnect();

    if (restart_delay_secs > 0)
    {
      g_eventLogger->info("Delaying Ndb restart for %u seconds.",
                          restart_delay_secs);
      NdbSleep_SecSleep(restart_delay_secs);
    };

    const int connnect_retries = 12;
    const int connect_delay = 5;
    const int verbose = 1;
    if (retriever.do_connect(connnect_retries, connect_delay, verbose) != 0)
    {
      g_eventLogger->error("Could not connect to management server, "
                           "error: '%s'", retriever.getErrorString());
      angel_exit(1);
    }
    g_eventLogger->info("Angel reconnected to '%s:%d'",
                        retriever.get_mgmd_host(),
                        retriever.get_mgmd_port());

    // Tell retriver to allocate the same nodeid again
    retriever.setNodeId(nodeid);

    g_eventLogger->debug("Angel reallocating nodeid %d", nodeid);
    const int alloc_retries = 10;
    const int alloc_delay = 3;
    const Uint32 realloced = retriever.allocNodeId(alloc_retries, alloc_delay);
    if (realloced == 0)
    {
      g_eventLogger->error("Angel failed to allocate nodeid, error: '%s'",
                           retriever.getErrorString());
      angel_exit(1);
    }
    if (realloced != nodeid)
    {
      g_eventLogger->error("Angel failed to reallocate nodeid %d, got %d",
                           nodeid, realloced);
      angel_exit(1);
    }
    g_eventLogger->info("Angel reallocated nodeid: %u", nodeid);

  }

  abort(); // Never reached
}


/*
  Order angel to shutdown it's ndbd
*/
void angel_stop(void)
{
  stop_child = true;
}