mgmapi.cpp

来自「MySQL源码文件5.X系列, 可自已编译到服务器」· C++ 代码 · 共 2,561 行 · 第 1/5 页

  while (sockfd == NDB_INVALID_SOCKET)
  {
    // do all the mgmt servers
    for (i = 0; i < cfg.ids.size(); i++)
    {
      if (cfg.ids[i].type != MgmId_TCP)
	continue;
      sockfd = s.connect(cfg.ids[i].name.c_str(), cfg.ids[i].port);
      if (sockfd != NDB_INVALID_SOCKET)
	break;
    }
    if (sockfd != NDB_INVALID_SOCKET)
      break;
#ifndef DBUG_OFF
    {
      DBUG_PRINT("info",("Unable to connect with connect string: %s",
			 cfg.makeConnectString(buf,sizeof(buf))));
    }
#endif
    if (verbose > 0) {
      fprintf(handle->errstream, 
	      "Unable to connect with connect string: %s\n",
	      cfg.makeConnectString(buf,sizeof(buf)));
      verbose= -1;
    }
    if (no_retries == 0) {
      setError(handle, NDB_MGM_COULD_NOT_CONNECT_TO_SOCKET, __LINE__,
	       "Unable to connect with connect string: %s",
	       cfg.makeConnectString(buf,sizeof(buf)));
      if (verbose == -2)
	fprintf(handle->errstream, ", failed.\n");
      DBUG_RETURN(-1);
    }
    if (verbose == -1) {
      fprintf(handle->errstream, "Retrying every %d seconds", 
	      retry_delay_in_seconds);
      if (no_retries > 0)
	fprintf(handle->errstream, ". Attempts left:");
      else
	fprintf(handle->errstream, ", until connected.");
      fflush(handle->errstream);
      verbose= -2;
    }
    if (no_retries > 0) {
      if (verbose == -2) {
	fprintf(handle->errstream, " %d", no_retries);
	fflush(handle->errstream);
      }
      no_retries--;
    }
    NdbSleep_SecSleep(retry_delay_in_seconds);
  }
  if (verbose == -2)
  {
    fprintf(handle->errstream, "\n");
    fflush(handle->errstream);
  }
  handle->cfg_i = i;
  
  handle->socket    = sockfd;
  handle->connected = 1;

  DBUG_RETURN(0);
}

/**
 * Only used for low level testing
 * Never to be used by end user.
 * Or anybody who doesn't know exactly what they're doing.
 */
extern "C"
int
ndb_mgm_get_fd(NdbMgmHandle handle)
{
  return handle->socket;
}

/**
 * Disconnect from a mgm server
 */
extern "C"
int
ndb_mgm_disconnect(NdbMgmHandle handle)
{
  SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_disconnect");
  CHECK_HANDLE(handle, -1);
  CHECK_CONNECTED(handle, -1);

  NDB_CLOSE_SOCKET(handle->socket);
  handle->socket = NDB_INVALID_SOCKET;
  handle->connected = 0;

  return 0;
}

struct ndb_mgm_type_atoi 
{
  const char * str;
  const char * alias;
  enum ndb_mgm_node_type value;
};

static struct ndb_mgm_type_atoi type_values[] = 
{
  { "NDB", "ndbd", NDB_MGM_NODE_TYPE_NDB},
  { "API", "mysqld", NDB_MGM_NODE_TYPE_API },
  { "MGM", "ndb_mgmd", NDB_MGM_NODE_TYPE_MGM }
};

const int no_of_type_values = (sizeof(type_values) / 
			       sizeof(ndb_mgm_type_atoi));

extern "C"
ndb_mgm_node_type
ndb_mgm_match_node_type(const char * type)
{
  if(type == 0)
    return NDB_MGM_NODE_TYPE_UNKNOWN;
  
  for(int i = 0; i<no_of_type_values; i++)
    if(strcmp(type, type_values[i].str) == 0)
      return type_values[i].value;
    else if(strcmp(type, type_values[i].alias) == 0)
      return type_values[i].value;
  
  return NDB_MGM_NODE_TYPE_UNKNOWN;
}

extern "C"
const char * 
ndb_mgm_get_node_type_string(enum ndb_mgm_node_type type)
{
  for(int i = 0; i<no_of_type_values; i++)
    if(type_values[i].value == type)
      return type_values[i].str;
  return 0;
}

extern "C"
const char * 
ndb_mgm_get_node_type_alias_string(enum ndb_mgm_node_type type, const char** str)
{
  for(int i = 0; i<no_of_type_values; i++)
    if(type_values[i].value == type)
      {
	if (str)
	  *str= type_values[i].str;
	return type_values[i].alias;
      }
  return 0;
}

struct ndb_mgm_status_atoi {
  const char * str;
  enum ndb_mgm_node_status value;
};

static struct ndb_mgm_status_atoi status_values[] = 
{
  { "UNKNOWN", NDB_MGM_NODE_STATUS_UNKNOWN },
  { "NO_CONTACT", NDB_MGM_NODE_STATUS_NO_CONTACT },
  { "NOT_STARTED", NDB_MGM_NODE_STATUS_NOT_STARTED },
  { "STARTING", NDB_MGM_NODE_STATUS_STARTING },
  { "STARTED", NDB_MGM_NODE_STATUS_STARTED },
  { "SHUTTING_DOWN", NDB_MGM_NODE_STATUS_SHUTTING_DOWN },
  { "RESTARTING", NDB_MGM_NODE_STATUS_RESTARTING },
  { "SINGLE USER MODE", NDB_MGM_NODE_STATUS_SINGLEUSER }
};

const int no_of_status_values = (sizeof(status_values) / 
				 sizeof(ndb_mgm_status_atoi));

extern "C"
ndb_mgm_node_status
ndb_mgm_match_node_status(const char * status)
{
  if(status == 0)
    return NDB_MGM_NODE_STATUS_UNKNOWN;
  
  for(int i = 0; i<no_of_status_values; i++)
    if(strcmp(status, status_values[i].str) == 0)
      return status_values[i].value;

  return NDB_MGM_NODE_STATUS_UNKNOWN;
}

extern "C"
const char * 
ndb_mgm_get_node_status_string(enum ndb_mgm_node_status status)
{
  int i;
  for(i = 0; i<no_of_status_values; i++)
    if(status_values[i].value == status)
      return status_values[i].str;

  for(i = 0; i<no_of_status_values; i++)
    if(status_values[i].value == NDB_MGM_NODE_STATUS_UNKNOWN)
      return status_values[i].str;
  
  return 0;
}

static int
status_ackumulate(struct ndb_mgm_node_state * state,
		  const char * field,
		  const char * value)
{
  if(strcmp("type", field) == 0){
    state->node_type = ndb_mgm_match_node_type(value);
  } else if(strcmp("status", field) == 0){
    state->node_status = ndb_mgm_match_node_status(value);
  } else if(strcmp("startphase", field) == 0){
    state->start_phase = atoi(value);
  } else if(strcmp("dynamic_id", field) == 0){
    state->dynamic_id = atoi(value);
  } else if(strcmp("node_group", field) == 0){
    state->node_group = atoi(value);
  } else if(strcmp("version", field) == 0){
    state->version = atoi(value);
  } else if(strcmp("connect_count", field) == 0){
    state->connect_count = atoi(value);    
  } else if(strcmp("address", field) == 0){
    strncpy(state->connect_address, value, sizeof(state->connect_address));
    state->connect_address[sizeof(state->connect_address)-1]= 0;
  } else {
    ndbout_c("Unknown field: %s", field);
  }
  return 0;
}

/**
 * Compare function for qsort() that sorts ndb_mgm_node_state in
 * node_id order
 */
static int
cmp_state(const void *_a, const void *_b) 
{
  struct ndb_mgm_node_state *a, *b;

  a = (struct ndb_mgm_node_state *)_a;
  b = (struct ndb_mgm_node_state *)_b;

  if (a->node_id > b->node_id)
    return 1;
  return -1;
}

extern "C"
struct ndb_mgm_cluster_state * 
ndb_mgm_get_status(NdbMgmHandle handle)
{
  SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_get_status");
  CHECK_HANDLE(handle, NULL);
  CHECK_CONNECTED(handle, NULL);

  SocketOutputStream out(handle->socket);
  SocketInputStream in(handle->socket, handle->read_timeout);

  out.println("get status");
  out.println("");

  char buf[1024];
  if(!in.gets(buf, sizeof(buf)))
  {
    SET_ERROR(handle, NDB_MGM_ILLEGAL_SERVER_REPLY, "Probably disconnected");
    return NULL;
  }
  if(strcmp("node status\n", buf) != 0) {
    SET_ERROR(handle, NDB_MGM_ILLEGAL_NODE_STATUS, buf);
    return NULL;
  }
  if(!in.gets(buf, sizeof(buf)))
  {
    SET_ERROR(handle, NDB_MGM_ILLEGAL_SERVER_REPLY, "Probably disconnected");
    return NULL;
  }

  BaseString tmp(buf);
  Vector<BaseString> split;
  tmp.split(split, ":");
  if(split.size() != 2){
    SET_ERROR(handle, NDB_MGM_ILLEGAL_NODE_STATUS, buf);
    return NULL;
  }

  if(!(split[0].trim() == "nodes")){
    SET_ERROR(handle, NDB_MGM_ILLEGAL_NODE_STATUS, buf);
    return NULL;
  }

  const int noOfNodes = atoi(split[1].c_str());

  ndb_mgm_cluster_state *state = (ndb_mgm_cluster_state*)
    malloc(sizeof(ndb_mgm_cluster_state)+
	   noOfNodes*(sizeof(ndb_mgm_node_state)+sizeof("000.000.000.000#")));

  if(!state)
  {
    SET_ERROR(handle, NDB_MGM_OUT_OF_MEMORY,
              "Allocating ndb_mgm_cluster_state");
    return NULL;
  }

  state->no_of_nodes= noOfNodes;
  ndb_mgm_node_state * ptr = &state->node_states[0];
  int nodeId = 0;
  int i;
  for (i= 0; i < noOfNodes; i++) {
    state->node_states[i].connect_address[0]= 0;
  }
  i = -1; ptr--;
  for(; i<noOfNodes; ){
    if(!in.gets(buf, sizeof(buf)))
    {
      free(state);
      SET_ERROR(handle, NDB_MGM_ILLEGAL_SERVER_REPLY,
                "Probably disconnected");
      return NULL;
    }
    tmp.assign(buf);

    if(tmp.trim() == ""){
      break;
    }
    
    Vector<BaseString> split;
    tmp.split(split, ":.", 4);
    if(split.size() != 4)
      break;
    
    const int id = atoi(split[1].c_str());
    if(id != nodeId){
      ptr++;
      i++;
      nodeId = id;
      ptr->node_id = id;
    }

    split[3].trim(" \t\n");

    if(status_ackumulate(ptr,split[2].c_str(), split[3].c_str()) != 0) {
      break;
    }
  }

  if(i+1 != noOfNodes){
    free(state);
    SET_ERROR(handle, NDB_MGM_ILLEGAL_NODE_STATUS, "Node count mismatch");
    return NULL;
  }

  qsort(state->node_states, state->no_of_nodes, sizeof(state->node_states[0]),
	cmp_state);
  return state;
}

extern "C"
int 
ndb_mgm_enter_single_user(NdbMgmHandle handle,
			  unsigned int nodeId,
			  struct ndb_mgm_reply* /*reply*/) 
{
  SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_enter_single_user");
  const ParserRow<ParserDummy> enter_single_reply[] = {
    MGM_CMD("enter single user reply", NULL, ""),
    MGM_ARG("result", String, Mandatory, "Error message"),
    MGM_END()
  };
  CHECK_HANDLE(handle, -1);
  CHECK_CONNECTED(handle, -1);

  Properties args;
  args.put("nodeId", nodeId);
  const Properties *reply;
  reply = ndb_mgm_call(handle, enter_single_reply, "enter single user", &args);
  CHECK_REPLY(reply, -1);

  BaseString result;
  reply->get("result", result);
  if(strcmp(result.c_str(), "Ok") != 0) {
    SET_ERROR(handle, NDB_MGM_COULD_NOT_ENTER_SINGLE_USER_MODE, 
	      result.c_str());
    delete reply;
    return -1;
  }

  delete reply;
  return 0;
}


extern "C"
int 
ndb_mgm_exit_single_user(NdbMgmHandle handle, struct ndb_mgm_reply* /*reply*/) 
{
  SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_exit_single_user");
  const ParserRow<ParserDummy> exit_single_reply[] = {
    MGM_CMD("exit single user reply", NULL, ""),
    MGM_ARG("result", String, Mandatory, "Error message"),
    MGM_END()
  };
  CHECK_HANDLE(handle, -1);
  CHECK_CONNECTED(handle, -1);

  const Properties *reply;
  reply = ndb_mgm_call(handle, exit_single_reply, "exit single user", 0);
  CHECK_REPLY(reply, -1);

  const char * buf;
  reply->get("result", &buf);
  if(strcmp(buf,"Ok")!=0) {
    SET_ERROR(handle, NDB_MGM_COULD_NOT_EXIT_SINGLE_USER_MODE, buf);
    delete reply;    
    return -1;
  }

  delete reply;
  return 0;
}

extern "C"
int 
ndb_mgm_stop(NdbMgmHandle handle, int no_of_nodes, const int * node_list)
{
  SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_stop");
  return ndb_mgm_stop2(handle, no_of_nodes, node_list, 0);
}

extern "C"
int
ndb_mgm_stop2(NdbMgmHandle handle, int no_of_nodes, const int * node_list,
	      int abort)
{
  int disconnect;
  return ndb_mgm_stop3(handle, no_of_nodes, node_list, abort, &disconnect);
}


extern "C"
int
ndb_mgm_stop3(NdbMgmHandle handle, int no_of_nodes, const int * node_list,
	      int abort, int *disconnect)
{
  SET_ERROR(handle, NDB_MGM_NO_ERROR, "Executing: ndb_mgm_stop3");
  const ParserRow<ParserDummy> stop_reply_v1[] = {
    MGM_CMD("stop reply", NULL, ""),
    MGM_ARG("stopped", Int, Optional, "No of stopped nodes"),
    MGM_ARG("result", String, Mandatory, "Error message"),
    MGM_END()
  };
  const ParserRow<ParserDummy> stop_reply_v2[] = {
    MGM_CMD("stop reply", NULL, ""),
    MGM_ARG("stopped", Int, Optional, "No of stopped nodes"),
    MGM_ARG("result", String, Mandatory, "Error message"),
    MGM_ARG("disconnect", Int, Mandatory, "Need to disconnect"),
    MGM_END()
  };

  CHECK_HANDLE(handle, -1);
  CHECK_CONNECTED(handle, -1);

  if(handle->mgmd_version_build==-1)
  {
    char verstr[50];
    if(!ndb_mgm_get_version(handle,
                        &(handle->mgmd_version_major),
                        &(handle->mgmd_version_minor),
                        &(handle->mgmd_version_build),
                        sizeof(verstr),
                            verstr))
    {
      return -1;
    }
  }
  int use_v2= ((handle->mgmd_version_major==5)
    && (
        (handle->mgmd_version_minor==0 && handle->mgmd_version_build>=21)
        ||(handle->mgmd_version_minor==1 && handle->mgmd_version_build>=12)
        ||(handle->mgmd_version_minor>1)
        )
               )
    || (handle->mgmd_version_major>5);

  if(no_of_nodes < -1){
    SET_ERROR(handle, NDB_MGM_ILLEGAL_NUMBER_OF_NODES, 
	      "Negative number of nodes requested to stop");
    return -1;
  }

  Uint32 stoppedNoOfNodes = 0;
  if(no_of_nodes <= 0){
    /**
     * All nodes should be stopped (all or just db)
     */
    Properties args;
    args.put("abort", abort);
    if(use_v2)
      args.put("stop", (no_of_nodes==-1)?"mgm,db":"db");
    const Properties *reply;
    if(use_v2)
      reply = ndb_mgm_call(handle, stop_reply_v2, "stop all", &args);
    else
      reply = ndb_mgm_call(handle, stop_reply_v1, "stop all", &args);
    CHECK_REPLY(reply, -1);

    if(!reply->get("stopped", &stoppedNoOfNodes)){
      SET_ERROR(handle, NDB_MGM_STOP_FAILED, 
		"Could not get number of stopped nodes from mgm server");
      delete reply;
      return -1;
    }
    if(use_v2)
      reply->get("disconnect", (Uint32*)disconnect);