messages.c

linux集群服务器软件代码包

		s_crmd_fsa(C_IPC_MESSAGE);
		
		return can_reply;
	}
	
	crm_devel("received client join msg");
	crm_log_message(LOG_MSG, client_msg->msg);
	auth_result = process_hello_message(
		client_msg->xml, &uuid, &client_name,
		&major_version, &minor_version);

	if (auth_result == TRUE) {
		if(client_name == NULL || uuid == NULL) {
			crm_err("Bad client details (client_name=%s, uuid=%s)",
				crm_str(client_name), crm_str(uuid));
			auth_result = FALSE;
		}
	}

	if (auth_result == TRUE) {
		/* check version */
		int mav = atoi(major_version);
		int miv = atoi(minor_version);
		crm_devel("Checking client version number");
		if (mav < 0 || miv < 0) {
			crm_err("Client version (%d:%d) is not acceptable",
				mav, miv);
			auth_result = FALSE;
		}
		crm_free(major_version);
		crm_free(minor_version);
	}

	if (auth_result == TRUE) {
		/* if we already have one of those clients
		 * only applies to te, pe etc.  not admin clients
		 */

		if (strcmp(CRM_SYSTEM_PENGINE, client_name) == 0) {
			the_subsystem = pe_subsystem;
			
		} else if (strcmp(CRM_SYSTEM_TENGINE, client_name) == 0) {
			the_subsystem = te_subsystem;
		}

		if (the_subsystem != NULL) {
			/* do we already have one? */
			crm_devel("Checking if %s is required/already connected",
				  client_name);
			
			if(is_set(fsa_input_register,
				  the_subsystem->flag_connected)) {
				auth_result = FALSE;
				crm_err("Bit\t%.16llx set in %.16llx",
					  the_subsystem->flag_connected,
					  fsa_input_register);
				crm_err("Client %s is already connected",
					client_name);
			} else if(FALSE == is_set(fsa_input_register,
					 the_subsystem->flag_required)) {
				auth_result = FALSE;
				crm_err("Bit\t%.16llx not set in %.16llx",
					  the_subsystem->flag_connected,
					  fsa_input_register);
				crm_err("Client %s joined without us asking",
					client_name);
			} else {
				the_subsystem->ipc =
					curr_client->client_channel;
			}

		} else {
			table_key = (gpointer)
				generate_hash_key(client_name, uuid);
		}
	}
	
	if (auth_result == TRUE) {
		if(table_key == NULL) {
			table_key = (gpointer)crm_strdup(client_name);
		}
		crm_info("Accepted client %s", crm_str(table_key));

		curr_client->table_key = table_key;
		curr_client->sub_sys = crm_strdup(client_name);
		curr_client->uuid = crm_strdup(uuid);
	
		g_hash_table_insert (ipc_clients,
				     table_key,
				     curr_client->client_channel);

		send_hello_message(curr_client->client_channel,
				   "n/a", CRM_SYSTEM_CRMD,
				   "0", "1");

		crm_debug("Updated client list with %s", crm_str(table_key));
		
		if(the_subsystem != NULL) {
			set_bit_inplace(
				fsa_input_register, the_subsystem->flag_connected);
		}
		s_crmd_fsa(C_SUBSYSTEM_CONNECT);

	} else {
		crm_err("Rejected client logon request");
		curr_client->client_channel->ch_status = IPC_DISC_PENDING;
	}
	
	if(uuid != NULL) crm_free(uuid);
	if(minor_version != NULL) crm_free(minor_version);
	if(major_version != NULL) crm_free(major_version);
	if(client_name != NULL) crm_free(client_name);

	/* hello messages should never be processed further */
	return FALSE;
}

enum crmd_fsa_input
handle_message(ha_msg_input_t *stored_msg)
{
	enum crmd_fsa_input next_input = I_NULL;
	const char *type = NULL;
	if(stored_msg == NULL || stored_msg->msg == NULL) {
		crm_err("No message to handle");
		return I_NULL;
	}
	type = cl_get_string(stored_msg->msg, F_CRM_MSG_TYPE);

	if(safe_str_eq(type, XML_ATTR_REQUEST)) {
		next_input = handle_request(stored_msg);

	} else if(safe_str_eq(type, XML_ATTR_RESPONSE)) {
		next_input = handle_response(stored_msg);

	} else {
		crm_err("Unknown message type: %s", type);
	}

/* 	crm_verbose("%s: Next input is %s", __FUNCTION__, */
/* 		   fsa_input2string(next_input)); */
	
	return next_input;
}


enum crmd_fsa_input
handle_request(ha_msg_input_t *stored_msg)
{
	HA_Message *msg = NULL;
	enum crmd_fsa_input next_input = I_NULL;

	const char *op        = cl_get_string(stored_msg->msg, F_CRM_TASK);
	const char *sys_to    = cl_get_string(stored_msg->msg, F_CRM_SYS_TO);
	const char *host_from = cl_get_string(stored_msg->msg, F_CRM_HOST_FROM);

	crm_verbose("Received %s in state %s", op, fsa_state2string(fsa_state));
	
	if(op == NULL) {
		crm_err("Bad message");
		crm_log_message(LOG_ERR, stored_msg->msg);

		/*========== common actions ==========*/
	} else if(strcmp(op, CRM_OP_VOTE) == 0) {
		/* count the vote and decide what to do after that */
		register_fsa_input_w_actions(
			C_HA_MESSAGE, I_NULL, stored_msg, A_ELECTION_COUNT);

		/* Sometimes we _must_ go into S_ELECTION */
		if(fsa_state == S_HALT) {
			crm_debug("Forcing an election from S_HALT");
			next_input = I_ELECTION;
#if 0
		} else if(AM_I_DC) {
		/* This is the old way of doing things but what is gained? */
			next_input = I_ELECTION;
#endif
		}
		
	} else if(strcmp(op, CRM_OP_LOCAL_SHUTDOWN) == 0) {
		
		crm_shutdown(SIGTERM);
		/*next_input = I_SHUTDOWN; */
		next_input = I_NULL;
			
	} else if(strcmp(op, CRM_OP_PING) == 0) {
		/* eventually do some stuff to figure out
		 * if we /are/ ok
		 */
		crm_data_t *ping = createPingAnswerFragment(sys_to, "ok");

		set_xml_property_copy(ping, "crmd_state",
				      fsa_state2string(fsa_state));
		
		msg = create_reply(stored_msg->msg, ping);
		
		if(relay_message(msg, TRUE) == FALSE) {
			crm_msg_del(msg);
		}
		
		/* probably better to do this via signals on the
		 * local node
		 */
	} else if(strcmp(op, CRM_OP_DEBUG_UP) == 0) {
		int level = get_crm_log_level();
		set_crm_log_level(level+1);
		crm_info("Debug set to %d (was %d)",
			 get_crm_log_level(), level);
		
	} else if(strcmp(op, CRM_OP_DEBUG_DOWN) == 0) {
		int level = get_crm_log_level();
		set_crm_log_level(level-1);
		crm_info("Debug set to %d (was %d)",
			 get_crm_log_level(), level);

	} else if(strcmp(op, CRM_OP_WELCOME) == 0) {
		next_input = I_JOIN_OFFER;
				
	} else if(strcmp(op, CRM_OP_JOINACK) == 0) {
		next_input = I_JOIN_RESULT;

		/* this functionality should only be enabled if this is a development build */
	} else if(CRM_DEV_BUILD && strcmp(op, CRM_OP_DIE) == 0/*constant condition*/) {
		crm_warn("Test-only code: Killing the CRM without mercy");
		crm_warn("Inhibiting respawns");
		exit(100);
		
		/*========== (NOT_DC)-Only Actions ==========*/
	} else if(AM_I_DC == FALSE){

		gboolean dc_match = safe_str_eq(host_from, fsa_our_dc);

		if(dc_match || fsa_our_dc == NULL) {
			if(strcmp(op, CRM_OP_HBEAT) == 0) {
				crm_devel("Received DC heartbeat from %s",
					  host_from);
				next_input = I_DC_HEARTBEAT;
				
			} else if(fsa_our_dc == NULL) {
				crm_warn("CRMd discarding request: %s"
					" (DC: %s, from: %s)",
					op, crm_str(fsa_our_dc), host_from);

				crm_warn("Ignored Request");
				crm_log_message(LOG_WARNING, stored_msg->msg);
				
			} else if(strcmp(op, CRM_OP_SHUTDOWN) == 0) {
				next_input = I_TERMINATE;
				
			} else {
				crm_err("CRMd didnt expect request: %s", op);
				crm_log_message(LOG_ERR, stored_msg->msg);
			}
			
		} else {
			crm_warn("Discarding %s op from %s", op, host_from);
		}

		/*========== DC-Only Actions ==========*/
	} else if(AM_I_DC){
		if(strcmp(op, CRM_OP_TEABORT) == 0) {
			if(fsa_state != S_INTEGRATION) {
				next_input = I_PE_CALC;

			} else {	
				crm_devel("Ignoring %s in state %s."
					"  Waiting for the integration to"
					" complete first.",
					op, fsa_state2string(fsa_state));
			}
				
		} else if(strcmp(op, CRM_OP_TECOMPLETE) == 0) {
/* 			if(fsa_state == S_TRANSITION_ENGINE) { */
				next_input = I_TE_SUCCESS;
/* 			} else { */
/* 				crm_warn("Op %s is only valid in state %s..." */
/* 					 "We are in (%s)", */
/* 					 op, */
/* 					 fsa_state2string(S_TRANSITION_ENGINE), */
/* 					 fsa_state2string(fsa_state)); */
/* 			} */

		} else if(strcmp(op, CRM_OP_ANNOUNCE) == 0) {
			next_input = I_NODE_JOIN;
			
		} else if(strcmp(op, CRM_OP_SHUTDOWN) == 0) {
			gboolean dc_match = safe_str_eq(host_from, fsa_our_dc);
			if(dc_match) {
				crm_err("We didnt ask to be shut down yet our"
					" TE is telling us too."
					" Better get out now!");
				next_input = I_TERMINATE;

			} else if(is_set(fsa_input_register, R_SHUTDOWN)) {
				crm_err("We asked to be shut down, "
					" are still the DC, yet another node"
					" is askin us to shutdown!");
				next_input = I_TERMINATE;			

			} else if(fsa_state != S_STOPPING) {
				crm_warn("Another node is asking us to shutdown"
					" but we think we're ok.");
				next_input = I_ELECTION;			
			}
			
		} else if(strcmp(op, CRM_OP_SHUTDOWN_REQ) == 0) {
			/* a slave wants to shut down */
			/* create cib fragment and add to message */
			next_input = handle_shutdown_request(stored_msg->msg);
			
		} else {
			crm_err("Unexpected request (%s) sent to the DC", op);
			crm_log_message(LOG_ERR, stored_msg->msg);
		}		
	}
	return next_input;
}

enum crmd_fsa_input
handle_response(ha_msg_input_t *stored_msg)
{
	enum crmd_fsa_input next_input = I_NULL;

	const char *op        = cl_get_string(stored_msg->msg, F_CRM_TASK);
	const char *sys_from  = cl_get_string(stored_msg->msg, F_CRM_SYS_FROM);
	const char *msg_ref   = cl_get_string(stored_msg->msg, XML_ATTR_REFERENCE);

	crm_verbose("Received %s %s in state %s",
		    op, XML_ATTR_RESPONSE, fsa_state2string(fsa_state));
	
	if(op == NULL) {
		crm_err("Bad message");
		crm_log_message(LOG_ERR, stored_msg->msg);

	} else if(AM_I_DC && strcmp(op, CRM_OP_WELCOME) == 0) {
		next_input = I_JOIN_REQUEST;
				
	} else if(AM_I_DC && strcmp(op, CRM_OP_JOINACK) == 0) {
		next_input = I_JOIN_RESULT;
				
 	} else if(AM_I_DC && strcmp(op, CRM_OP_PECALC) == 0) {

		if(safe_str_eq(msg_ref, fsa_pe_ref)) {
			next_input = I_PE_SUCCESS;
			
		} else {
			crm_verbose("Skipping superceeded reply from %s",
				    sys_from);
		}
		
	} else if(strcmp(op, CRM_OP_VOTE) == 0
		  || strcmp(op, CRM_OP_HBEAT) == 0
		  || strcmp(op, CRM_OP_WELCOME) == 0
		  || strcmp(op, CRM_OP_SHUTDOWN_REQ) == 0
		  || strcmp(op, CRM_OP_SHUTDOWN) == 0
		  || strcmp(op, CRM_OP_ANNOUNCE) == 0) {
		next_input = I_NULL;
		
	} else if(strcmp(op, CRM_OP_CIB_CREATE) == 0
		  || strcmp(op, CRM_OP_CIB_UPDATE) == 0
		  || strcmp(op, CRM_OP_CIB_DELETE) == 0
		  || strcmp(op, CRM_OP_CIB_REPLACE) == 0
		  || strcmp(op, CRM_OP_CIB_ERASE) == 0) {
		
		/* perhaps we should do somethign with these replies,
		 * especially check that the actions passed
		 */
/* 		fprintf(router_strm, "Message result: CIB Reply\n"); */

	} else {
		crm_err("Unexpected response (op=%s) sent to the %s",
			op, AM_I_DC?"DC":"CRMd");
		next_input = I_NULL;
	}
	
	return next_input;
		
}

enum crmd_fsa_input
handle_shutdown_request(HA_Message *stored_msg)
{
	/* handle here to avoid potential version issues
	 *   where the shutdown message/proceedure may have
	 *   been changed in later versions.
	 *
	 * This way the DC is always in control of the shutdown
	 */
	
	crm_data_t *frag = NULL;
	time_t now = time(NULL);
	char *now_s = crm_itoa((int)now);
	crm_data_t *node_state = create_xml_node(NULL, XML_CIB_TAG_STATE);
	const char *host_from= cl_get_string(stored_msg, F_CRM_HOST_FROM);
	
	crm_info("Creating shutdown request for %s",host_from);

	crm_info("stored msg");
	crm_log_message(LOG_MSG, stored_msg);
	
	set_uuid(fsa_cluster_conn, node_state, XML_ATTR_UUID, host_from);
	set_xml_property_copy(node_state, XML_ATTR_UNAME, host_from);
	set_xml_property_copy(node_state, XML_CIB_ATTR_SHUTDOWN,  now_s);
	set_xml_property_copy(
		node_state, XML_CIB_ATTR_EXPSTATE, CRMD_STATE_INACTIVE);
	
	frag = create_cib_fragment(node_state, NULL);
	
	/* cleanup intermediate steps */
	free_xml(node_state);
	crm_free(now_s);

	fsa_cib_conn->cmds->modify(
		fsa_cib_conn, XML_CIB_TAG_STATUS, frag, NULL,
		cib_sync_call);

	free_xml(frag);

	/* will be picked up by the TE as long as its running */
	if(! is_set(fsa_input_register, R_TE_CONNECTED) ) {
		register_fsa_input(C_HA_MESSAGE, I_PE_CALC, NULL);
	}
	return I_NULL;
}

/* frees msg upon completion */
gboolean
send_msg_via_ha(ll_cluster_t *hb_fd, HA_Message *msg)
{
	int log_level        = LOG_DEV;
	int send_result      = HA_OK;
	gboolean broadcast   = FALSE;
	gboolean all_is_good = TRUE;

	const char *op       = cl_get_string(msg, F_CRM_TASK);
	const char *sys_to   = cl_get_string(msg, F_CRM_SYS_TO);
	const char *host_to  = cl_get_string(msg, F_CRM_HOST_TO);

	if (msg == NULL) {
		crm_err("Attempt to send NULL Message via HA failed.");
		all_is_good = FALSE;
	} else {
		crm_trace("Relaying message to (%s) via HA", host_to); 
	}
	
	if (all_is_good) {
		if (sys_to == NULL || strlen(sys_to) == 0) {
			crm_err("You did not specify a destination sub-system"
				" for this message.");
			all_is_good = FALSE;
		}
	}

	/* There are a number of messages may not need to be ordered.
	 * At a later point perhaps we should detect them and send them
	 *  as unordered messages.
	 */
	if (all_is_good) {
		if (host_to == NULL
		    || strlen(host_to) == 0
		    || safe_str_eq(sys_to, CRM_SYSTEM_DC)) {
			broadcast = TRUE;
			send_result=hb_fd->llc_ops->sendclustermsg(hb_fd, msg);
		} else {
			send_result = hb_fd->llc_ops->send_ordered_nodemsg(
				hb_fd, msg, host_to);
		}
		
		if(send_result != HA_OK)
			all_is_good = FALSE;
	}
	
	if(all_is_good == FALSE) {
		log_level = LOG_ERR;
	}

	if(log_level == LOG_ERR
	   || (safe_str_neq(op, CRM_OP_HBEAT))) {
		do_crm_log(log_level, __FUNCTION__, NULL,
			   "Sending %sHA message (ref=%s) to %s@%s %s.",
			   broadcast?"broadcast ":"directed ",
			   cl_get_string(msg, XML_ATTR_REFERENCE),
			   crm_str(sys_to), host_to==NULL?"<all>":host_to,
			   all_is_good?"succeeded":"failed");
	}
	
#ifdef MSG_LOG
	crm_log_message_adv(log_level, "outbound.log", msg);
#endif

	crm_msg_del(msg);
	
	return all_is_good;
}


/* msg is deleted by the time this returns */

gboolean
send_msg_via_ipc(HA_Message *msg, const char *sys)
{
	gboolean send_ok = TRUE;
	IPC_Channel *client_channel;
	enum crmd_fsa_input next_input;
	
	crm_trace("relaying msg to sub_sys=%s via IPC", sys);

	client_channel =
		(IPC_Channel*)g_hash_table_lookup(ipc_clients, sys);

	if(cl_get_string(msg, F_CRM_HOST_FROM) == NULL) {
		ha_msg_add(msg, F_CRM_HOST_FROM, fsa_our_uname);
	}
	
	if (client_channel != NULL) {
		crm_devel("Sending message via channel %s.", sys);
		send_ok = send_ipc_message(client_channel, msg);
		msg = NULL;  /* so the crm_msg_del() below doesnt fail */
		
	} else if(sys != NULL && strcmp(sys, CRM_SYSTEM_CIB) == 0) {
		crm_err("Sub-system (%s) has been incorporated into the CRMd.",
			sys);
		crm_err("Change the way we handle this CIB message");
		crm_log_message(LOG_ERR, msg);
		send_ok = FALSE;
		
	} else if(sys != NULL && strcmp(sys, CRM_SYSTEM_LRMD) == 0) {
		fsa_data_t *fsa_data = NULL;
		ha_msg_input_t *msg_copy = new_ha_msg_input(msg);

		crm_malloc(fsa_data, sizeof(fsa_data_t));
		fsa_data->fsa_input = I_MESSAGE;
		fsa_data->fsa_cause = C_IPC_MESSAGE;
		fsa_data->data = msg_copy;
		fsa_data->origin = __FUNCTION__;
		fsa_data->data_type = fsa_dt_ha_msg;
		
#ifdef FSA_TRACE
		crm_verbose("Invoking action %s (%.16llx)",
			    fsa_action2string(A_LRM_INVOKE),
			    A_LRM_INVOKE);
#endif
		next_input = do_lrm_invoke(A_LRM_INVOKE, C_IPC_MESSAGE,
					   fsa_state, I_MESSAGE, fsa_data);

		delete_ha_msg_input(msg_copy);
		crm_free(fsa_data);
		
		/* todo: feed this back in for anything != I_NULL */
		
#ifdef FSA_TRACE
		crm_verbose("Result of action %s was %s",
			    fsa_action2string(A_LRM_INVOKE),
			    fsa_input2string(next_input));
#endif
		
	} else {
		crm_err("Unknown Sub-system (%s)... discarding message.", sys);
		send_ok = FALSE;
	}    

	crm_msg_del(msg);

	return send_ok;
}