services.cpp

mysql-5.0.27版本源码包

/* Copyright (C) 2003 MySQL AB

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2 of the License, or
   (at your option) any later version.

   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 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#include <ndb_global.h>
#include <ctype.h>

#include <uucode.h>
#include <socket_io.h>
#include <ndb_version.h>
#include <mgmapi.h>
#include <EventLogger.hpp>
#include <signaldata/SetLogLevelOrd.hpp>
#include <LogLevel.hpp>
#include <BaseString.hpp>

#include <ConfigValues.hpp>
#include <mgmapi_configuration.hpp>
#include <Vector.hpp>
#include "Services.hpp"
#include "../mgmapi/ndb_logevent.hpp"

#include <base64.h>

extern bool g_StopServer;
extern bool g_RestartServer;
extern EventLogger g_eventLogger;

static const unsigned int MAX_READ_TIMEOUT = 1000 ;
static const unsigned int MAX_WRITE_TIMEOUT = 100 ;

/**
   const char * name;
   const char * realName;
   const Type type;
   const ArgType argType;
   const ArgRequired argRequired;
   const ArgMinMax argMinMax;
   const int minVal;
   const int maxVal;
   void (T::* function)(const class Properties & args);
   const char * description;
*/

#define MGM_CMD(name, fun, desc) \
 { name, \
   0, \
   ParserRow<MgmApiSession>::Cmd, \
   ParserRow<MgmApiSession>::String, \
   ParserRow<MgmApiSession>::Optional, \
   ParserRow<MgmApiSession>::IgnoreMinMax, \
   0, 0, \
   fun, \
   desc, 0 }

#define MGM_ARG(name, type, opt, desc) \
 { name, \
   0, \
   ParserRow<MgmApiSession>::Arg, \
   ParserRow<MgmApiSession>::type, \
   ParserRow<MgmApiSession>::opt, \
   ParserRow<MgmApiSession>::IgnoreMinMax, \
   0, 0, \
   0, \
  desc, 0 }

#define MGM_ARG2(name, type, opt, min, max, desc) \
 { name, \
   0, \
   ParserRow<MgmApiSession>::Arg, \
   ParserRow<MgmApiSession>::type, \
   ParserRow<MgmApiSession>::opt, \
   ParserRow<MgmApiSession>::IgnoreMinMax, \
   min, max, \
   0, \
  desc, 0 }

#define MGM_END() \
 { 0, \
   0, \
   ParserRow<MgmApiSession>::Arg, \
   ParserRow<MgmApiSession>::Int, \
   ParserRow<MgmApiSession>::Optional, \
   ParserRow<MgmApiSession>::IgnoreMinMax, \
   0, 0, \
   0, \
   0, 0 }

#define MGM_CMD_ALIAS(name, realName, fun) \
 { name, \
   realName, \
   ParserRow<MgmApiSession>::CmdAlias, \
   ParserRow<MgmApiSession>::Int, \
   ParserRow<MgmApiSession>::Optional, \
   ParserRow<MgmApiSession>::IgnoreMinMax, \
   0, 0, \
   0, \
   0, 0 }

#define MGM_ARG_ALIAS(name, realName, fun) \
 { name, \
   realName, \
   ParserRow<MgmApiSession>::ArgAlias, \
   ParserRow<MgmApiSession>::Int, \
   ParserRow<MgmApiSession>::Optional, \
   ParserRow<MgmApiSession>::IgnoreMinMax, \
   0, 0, \
   0, \
   0, 0 }

const
ParserRow<MgmApiSession> commands[] = {
  MGM_CMD("get statport", &MgmApiSession::getStatPort, ""),
  
  MGM_CMD("get config", &MgmApiSession::getConfig, ""),
    MGM_ARG("version", Int, Mandatory, "Configuration version number"),
    MGM_ARG("node", Int, Optional, "Node ID"),

  MGM_CMD("get nodeid", &MgmApiSession::get_nodeid, ""),
    MGM_ARG("version", Int, Mandatory, "Configuration version number"),
    MGM_ARG("nodetype", Int, Mandatory, "Node type"),
    MGM_ARG("transporter", String, Optional, "Transporter type"),
    MGM_ARG("nodeid", Int, Optional, "Node ID"),
    MGM_ARG("user", String, Mandatory, "Password"),
    MGM_ARG("password", String, Mandatory, "Password"),
    MGM_ARG("public key", String, Mandatory, "Public key"),
    MGM_ARG("endian", String, Optional, "Endianness"),
    MGM_ARG("name", String, Optional, "Name of connection"),
    MGM_ARG("timeout", Int, Optional, "Timeout in seconds"),
    MGM_ARG("log_event", Int, Optional, "Log failure in cluster log"),

  MGM_CMD("get version", &MgmApiSession::getVersion, ""),
  
  MGM_CMD("get status", &MgmApiSession::getStatus, ""),

  MGM_CMD("get info clusterlog", &MgmApiSession::getInfoClusterLog, ""),

  MGM_CMD("restart node", &MgmApiSession::restart_v1, ""),
    MGM_ARG("node", String, Mandatory, "Nodes to restart"),
    MGM_ARG("initialstart", Int, Optional, "Initial start"),
    MGM_ARG("nostart", Int, Optional, "No start"),
    MGM_ARG("abort", Int, Optional, "Abort"),

  MGM_CMD("restart node v2", &MgmApiSession::restart_v2, ""),
    MGM_ARG("node", String, Mandatory, "Nodes to restart"),
    MGM_ARG("initialstart", Int, Optional, "Initial start"),
    MGM_ARG("nostart", Int, Optional, "No start"),
    MGM_ARG("abort", Int, Optional, "Abort"),

  MGM_CMD("restart all", &MgmApiSession::restartAll, ""),
    MGM_ARG("initialstart", Int, Optional, "Initial start"),
    MGM_ARG("nostart", Int, Optional, "No start"),
    MGM_ARG("abort", Int, Optional, "Abort"),

  MGM_CMD("insert error", &MgmApiSession::insertError, ""),
    MGM_ARG("node", Int, Mandatory, "Node to receive error"),
    MGM_ARG("error", Int, Mandatory, "Errorcode to insert"),

  MGM_CMD("set trace", &MgmApiSession::setTrace, ""),
    MGM_ARG("node", Int, Mandatory, "Node"),
    MGM_ARG("trace", Int, Mandatory, "Trace number"),

  MGM_CMD("log signals", &MgmApiSession::logSignals, ""),
    MGM_ARG("node", Int, Mandatory, "Node"),
    MGM_ARG("blocks", String, Mandatory, "Blocks (space separated)"),
    MGM_ARG("in", Int, Mandatory, "Log input signals"),
    MGM_ARG("out", Int, Mandatory, "Log output signals"),

  MGM_CMD("start signallog", &MgmApiSession::startSignalLog, ""),
    MGM_ARG("node", Int, Mandatory, "Node"),

  MGM_CMD("stop signallog", &MgmApiSession::stopSignalLog, ""),
    MGM_ARG("node", Int, Mandatory, "Node"),

  MGM_CMD("dump state", &MgmApiSession::dumpState, ""),
    MGM_ARG("node", Int, Mandatory ,"Node"),
    MGM_ARG("args", String, Mandatory, "Args(space separated int's)"),

  MGM_CMD("start backup", &MgmApiSession::startBackup, ""),
    MGM_ARG("completed", Int, Optional ,"Wait until completed"),

  MGM_CMD("abort backup", &MgmApiSession::abortBackup, ""),
    MGM_ARG("id", Int, Mandatory, "Backup id"),

  MGM_CMD("stop", &MgmApiSession::stop_v1, ""),
    MGM_ARG("node", String, Mandatory, "Node"),
    MGM_ARG("abort", Int, Mandatory, "Node"),

  MGM_CMD("stop v2", &MgmApiSession::stop_v2, ""),
    MGM_ARG("node", String, Mandatory, "Node"),
    MGM_ARG("abort", Int, Mandatory, "Node"),

  MGM_CMD("stop all", &MgmApiSession::stopAll, ""),
    MGM_ARG("abort", Int, Mandatory, "Node"),
    MGM_ARG("stop", String, Optional, "MGM/DB or both"),

  MGM_CMD("enter single user", &MgmApiSession::enterSingleUser, ""),
    MGM_ARG("nodeId", Int, Mandatory, "Node"),
  
  MGM_CMD("exit single user", &MgmApiSession::exitSingleUser, ""),
  

  MGM_CMD("start", &MgmApiSession::start, ""),
    MGM_ARG("node", Int, Mandatory, "Node"),

  MGM_CMD("start all", &MgmApiSession::startAll, ""),

  MGM_CMD("bye", &MgmApiSession::bye, ""),

  MGM_CMD("end session", &MgmApiSession::endSession, ""),

  MGM_CMD("set loglevel", &MgmApiSession::setLogLevel, ""),
    MGM_ARG("node", Int, Mandatory, "Node"),
    MGM_ARG("category", Int, Mandatory, "Event category"),
    MGM_ARG("level", Int, Mandatory, "Log level (0-15)"),

  MGM_CMD("set cluster loglevel", &MgmApiSession::setClusterLogLevel, ""),
    MGM_ARG("node", Int, Mandatory, "Node"),
    MGM_ARG("category", Int, Mandatory, "Event category"),
    MGM_ARG("level", Int, Mandatory, "Log level (0-15)"),

  MGM_CMD("set logfilter", &MgmApiSession::setLogFilter, ""),
    MGM_ARG("level", Int, Mandatory, "Severety level"),
    MGM_ARG("enable", Int, Mandatory, "1=disable, 0=enable, -1=toggle"),

  MGM_CMD("set parameter", &MgmApiSession::setParameter, ""),
    MGM_ARG("node", String, Mandatory, "Node"),
    MGM_ARG("parameter", String, Mandatory, "Parameter"),
    MGM_ARG("value", String, Mandatory, "Value"),

  MGM_CMD("set connection parameter",
	  &MgmApiSession::setConnectionParameter, ""),
    MGM_ARG("node1", String, Mandatory, "Node1 ID"),
    MGM_ARG("node2", String, Mandatory, "Node2 ID"),
    MGM_ARG("param", String, Mandatory, "Parameter"),
    MGM_ARG("value", String, Mandatory, "Value"),

  MGM_CMD("get connection parameter",
	  &MgmApiSession::getConnectionParameter, ""),
    MGM_ARG("node1", String, Mandatory, "Node1 ID"),
    MGM_ARG("node2", String, Mandatory, "Node2 ID"),
    MGM_ARG("param", String, Mandatory, "Parameter"),

  MGM_CMD("listen event", &MgmApiSession::listen_event, ""),
    MGM_ARG("node", Int, Optional, "Node"),  
    MGM_ARG("parsable", Int, Optional, "Parsable"),  
    MGM_ARG("filter", String, Mandatory, "Event category"),

  MGM_CMD("purge stale sessions", &MgmApiSession::purge_stale_sessions, ""),

  MGM_CMD("check connection", &MgmApiSession::check_connection, ""),

  MGM_CMD("transporter connect", &MgmApiSession::transporter_connect, ""),

  MGM_CMD("get mgmd nodeid", &MgmApiSession::get_mgmd_nodeid, ""),

  MGM_CMD("report event", &MgmApiSession::report_event, ""),
    MGM_ARG("length", Int, Mandatory, "Length"),
    MGM_ARG("data", String, Mandatory, "Data"),

  MGM_END()
};

struct PurgeStruct
{
  NodeBitmask free_nodes;/* free nodes as reported
			  * by ndbd in apiRegReqConf
			  */
  BaseString *str;
  NDB_TICKS tick;
};

MgmApiSession::MgmApiSession(class MgmtSrvr & mgm, NDB_SOCKET_TYPE sock)
  : SocketServer::Session(sock), m_mgmsrv(mgm)
{
  DBUG_ENTER("MgmApiSession::MgmApiSession");
  m_input = new SocketInputStream(sock);
  m_output = new SocketOutputStream(sock);
  m_parser = new Parser_t(commands, *m_input, true, true, true);
  m_allocated_resources= new MgmtSrvr::Allocated_resources(m_mgmsrv);
  m_stopSelf= 0;
  DBUG_VOID_RETURN;
}

MgmApiSession::~MgmApiSession()
{
  DBUG_ENTER("MgmApiSession::~MgmApiSession");
  if (m_input)
    delete m_input;
  if (m_output)
    delete m_output;
  if (m_parser)
    delete m_parser;
  if (m_allocated_resources)
    delete m_allocated_resources;
  if(m_socket != NDB_INVALID_SOCKET)
  {
    NDB_CLOSE_SOCKET(m_socket);
    m_socket= NDB_INVALID_SOCKET;
  }
  if(m_stopSelf < 0)
    g_RestartServer= true;
  if(m_stopSelf)
    g_StopServer= true;
  DBUG_VOID_RETURN;
}

void
MgmApiSession::runSession()
{
  DBUG_ENTER("MgmApiSession::runSession");

  Parser_t::Context ctx;
  while(!m_stop) {
    m_parser->run(ctx, *this);

    if(ctx.m_currentToken == 0)
      break;

    switch(ctx.m_status) {
    case Parser_t::UnknownCommand:
#ifdef MGM_GET_CONFIG_BACKWARDS_COMPAT
      /* Backwards compatibility for old NDBs that still use
       * the old "GET CONFIG" command.
       */
	  size_t i;
      for(i=0; i<strlen(ctx.m_currentToken); i++)
	ctx.m_currentToken[i] = toupper(ctx.m_currentToken[i]);

      if(strncmp("GET CONFIG ", 
		 ctx.m_currentToken,
		 strlen("GET CONFIG ")) == 0)
	getConfig_old(ctx);
#endif /* MGM_GET_CONFIG_BACKWARDS_COMPAT */
      break;
    default:
      break;
    }
  }
  if(m_socket != NDB_INVALID_SOCKET)
  {
    NDB_CLOSE_SOCKET(m_socket);
    m_socket= NDB_INVALID_SOCKET;
  }

  DBUG_VOID_RETURN;
}

#ifdef MGM_GET_CONFIG_BACKWARDS_COMPAT
void
MgmApiSession::getConfig_old(Parser_t::Context &ctx) {
  Properties args;

  Uint32 version, node;

  if(sscanf(ctx.m_currentToken, "GET CONFIG %d %d",
	    (int *)&version, (int *)&node) != 2) {
    m_output->println("Expected 2 arguments for GET CONFIG");
    return;
  }

  /* Put arguments in properties object so we can call the real function */  
  args.put("version", version);
  args.put("node", node);
  getConfig_common(ctx, args, true);
}
#endif /* MGM_GET_CONFIG_BACKWARDS_COMPAT */

void
MgmApiSession::getConfig(Parser_t::Context &ctx, 
			 const class Properties &args) {
  getConfig_common(ctx, args);
}

static Properties *
backward(const char * base, const Properties* reply){
  Properties * ret = new Properties();
  Properties::Iterator it(reply);
  for(const char * name = it.first(); name != 0; name=it.next()){
    PropertiesType type;
    reply->getTypeOf(name, &type);
    switch(type){
    case PropertiesType_Uint32:{
      Uint32 val;
      reply->get(name, &val);
      ret->put(name, val);
    }
      break;
    case PropertiesType_char:
      {
	const char * val;
	reply->get(name, &val);
	ret->put(name, val);
	if(!strcmp(name, "Type") && !strcmp(val, "DB")){
	  ret->put("NoOfDiskBufferPages", (unsigned)0);
	  ret->put("NoOfDiskFiles", (unsigned)0);
	  ret->put("NoOfDiskClusters", (unsigned)0);
	  ret->put("NoOfFreeDiskClusters", (unsigned)0);
	  ret->put("NoOfDiskClustersPerDiskFile", (unsigned)0);
	  ret->put("NoOfConcurrentCheckpointsDuringRestart", (unsigned)1);
	  ret->put("NoOfConcurrentCheckpointsAfterRestart", (unsigned)1);
	  ret->put("NoOfConcurrentProcessesHandleTakeover", (unsigned)1);
	}
      }
      break;
    case PropertiesType_Properties:
      {
	const Properties * recurse;
	reply->get(name, &recurse);
	Properties * val = backward(name, recurse);
	ret->put(name, val);
      }
      break;
    case PropertiesType_Uint64:
      break;
    }
  }
  return ret;
}

void
MgmApiSession::get_nodeid(Parser_t::Context &,
			  const class Properties &args)
{
  const char *cmd= "get nodeid reply";
  Uint32 version, nodeid= 0, nodetype= 0xff;
  Uint32 timeout= 20;  // default seconds timeout
  const char * transporter;
  const char * user;
  const char * password;
  const char * public_key;
  const char * endian= NULL;
  const char * name= NULL;
  Uint32 log_event= 1;
  bool log_event_version;
  union { long l; char c[sizeof(long)]; } endian_check;

  args.get("version", &version);
  args.get("nodetype", &nodetype);
  args.get("transporter", &transporter);
  args.get("nodeid", &nodeid);
  args.get("user", &user);
  args.get("password", &password);
  args.get("public key", &public_key);
  args.get("endian", &endian);
  args.get("name", &name);
  args.get("timeout", &timeout);
  /* for backwards compatability keep track if client uses new protocol */
  log_event_version= args.get("log_event", &log_event);

  endian_check.l = 1;
  if(endian 
     && strcmp(endian,(endian_check.c[sizeof(long)-1])?"big":"little")!=0) {
    m_output->println(cmd);
    m_output->println("result: Node does not have the same endianness as the management server.");
    m_output->println("");
    return;
  }

  bool compatible;
  switch (nodetype) {
  case NODE_TYPE_MGM:
  case NODE_TYPE_API:
    compatible = ndbCompatible_mgmt_api(NDB_VERSION, version);
    break;
  case NODE_TYPE_DB:
    compatible = ndbCompatible_mgmt_ndb(NDB_VERSION, version);
    break;
  default:
    m_output->println(cmd);
    m_output->println("result: unknown nodetype %d", nodetype);
    m_output->println("");
    return;
  }

  struct sockaddr_in addr;
  SOCKET_SIZE_TYPE addrlen= sizeof(addr);
  int r = getpeername(m_socket, (struct sockaddr*)&addr, &addrlen);
  if (r != 0 ) {
    m_output->println(cmd);
    m_output->println("result: getpeername(%d) failed, err= %d", m_socket, r);
    m_output->println("");
    return;
  }

  NodeId tmp= nodeid;
  if(tmp == 0 || !m_allocated_resources->is_reserved(tmp)){
    BaseString error_string;
    int error_code;
    NDB_TICKS tick= 0;
    /* only report error on second attempt as not to clog the cluster log */
    while (!m_mgmsrv.alloc_node_id(&tmp, (enum ndb_mgm_node_type)nodetype, 
                                   (struct sockaddr*)&addr, &addrlen, error_code, error_string,
                                   tick == 0 ? 0 : log_event))
    {
      /* NDB_MGM_ALLOCID_CONFIG_MISMATCH is a non retriable error */
      if (tick == 0 && error_code != NDB_MGM_ALLOCID_CONFIG_MISMATCH)
      {
        // attempt to free any timed out reservations
        tick= NdbTick_CurrentMillisecond();
        struct PurgeStruct ps;
        m_mgmsrv.get_connected_nodes(ps.free_nodes);
        // invert connected_nodes to get free nodes
        ps.free_nodes.bitXORC(NodeBitmask());
        ps.str= 0;
        ps.tick= tick;
        m_mgmsrv.get_socket_server()->
          foreachSession(stop_session_if_timed_out,&ps);
	m_mgmsrv.get_socket_server()->checkSessions();
        error_string = "";
        continue;
      }
      const char *alias;
      const char *str;
      alias= ndb_mgm_get_node_type_alias_string((enum ndb_mgm_node_type)
						nodetype, &str);
      m_output->println(cmd);
      m_output->println("result: %s", error_string.c_str());
      /* only use error_code protocol if client knows about it */
      if (log_event_version)
        m_output->println("error_code: %d", error_code);
      m_output->println("");
      return;
    }
  }    
  
#if 0
  if (!compatible){
    m_output->println(cmd);
    m_output->println("result: incompatible version mgmt 0x%x and node 0x%x",
		      NDB_VERSION, version);
    m_output->println("");
    return;
  }
#endif
  
  m_output->println(cmd);
  m_output->println("nodeid: %u", tmp);
  m_output->println("result: Ok");
  m_output->println("");
  m_allocated_resources->reserve_node(tmp, timeout*1000);
  
  if (name)
    g_eventLogger.info("Node %d: %s", tmp, name);

  return;
}

void
MgmApiSession::getConfig_common(Parser_t::Context &,
				const class Properties &args,