messages.c

在LINUX下实现HA的源代码

	gboolean processing_complete = FALSE;
	const char *host_to = xmlGetProp(xml_relay_message, XML_ATTR_HOSTTO);
	const char *sys_to  = xmlGetProp(xml_relay_message, XML_ATTR_SYSTO);
	const char *sys_cc  = get_xml_attr(xml_relay_message, XML_TAG_OPTIONS,
					   XML_ATTR_SYSCC, FALSE);

	FNIN();

	if(xml_relay_message == NULL) {
		cl_log(LOG_ERR, "Cannot route empty message");
		FNRET(TRUE);
	}

	if(strcmp("hello", xml_relay_message->name) == 0) {
		/* quietly ignore */
		FNRET(TRUE);
	}

	if(strcmp(XML_MSG_TAG, xml_relay_message->name) != 0) {

		xml_message_debug(xml_relay_message,
				  "Bad message type, should be crm_message");
		cl_log(LOG_ERR, "Ignoring message of type %s",
		       xml_relay_message->name);
		FNRET(TRUE);
	}
	

	if(sys_to == NULL) {
		xml_message_debug(xml_relay_message,
				  "Message did not have any value for sys_to");
		cl_log(LOG_ERR, "Message did not have any value for %s",
		       XML_ATTR_SYSTO);
		FNRET(TRUE);
	}
	
	is_for_dc   = (strcmp(CRM_SYSTEM_DC,   sys_to) == 0);
	is_for_dcib = (strcmp(CRM_SYSTEM_DCIB, sys_to) == 0);
	is_for_cib  = (strcmp(CRM_SYSTEM_CIB,  sys_to) == 0);
	is_for_crm  = (strcmp(CRM_SYSTEM_CRMD, sys_to) == 0);
	
	is_local = 0;
	if(host_to == NULL || strlen(host_to) == 0) {
		if(is_for_dc)
			is_local = 0;
		else if(is_for_crm && originated_locally)
			is_local = 0;
		else
			is_local = 1;
		
	} else if(strcmp(fsa_our_uname, host_to) == 0) {
		is_local=1;
	}

#if 0
	CRM_DEBUG2("is_local    %d", is_local);
	CRM_DEBUG2("is_for_dcib %d", is_for_dcib);
	CRM_DEBUG2("is_for_dc   %d", is_for_dc);
	CRM_DEBUG2("is_for_crm  %d", is_for_crm);
	CRM_DEBUG2("AM_I_DC     %d", AM_I_DC);
	CRM_DEBUG2("sys_to      %s", sys_to);
	CRM_DEBUG2("host_to     %s", host_to);
#endif
/*
	if(AM_I_DC && sys_cc != NULL && strcmp(sys_cc, CRM_SYSTEM_DC) == 0) {
		dont_cc = FALSE;
	}
*/
	(void)sys_cc;
	
	if(is_for_dc || is_for_dcib) {
		if(AM_I_DC) {
			ROUTER_RESULT("Message result: DC/CRMd process");
			processing_complete = FALSE; /*  more to be done by caller */

		} else if(originated_locally) {
			ROUTER_RESULT("Message result: External relay to DC");
			send_msg_via_ha(xml_relay_message, NULL);
			processing_complete = TRUE; 

		} else {
			ROUTER_RESULT("Message result: Discard, not DC");
			processing_complete = TRUE; /*  discard */
		}
		
	} else if(is_local && (is_for_crm || is_for_cib)) {
		ROUTER_RESULT("Message result: CRMd process");

	} else if(is_local) {
		if(dont_cc) {
			ROUTER_RESULT("Message result: Local relay");
		} else {
			/* The DC should also get this message */
			ROUTER_RESULT("Message result: Local relay with CC");
		}
		send_msg_via_ipc(xml_relay_message, sys_to);
		processing_complete = TRUE & dont_cc;
		
	} else {
		if(dont_cc) {
			ROUTER_RESULT("Message result: External relay");
		} else {
			/* The DC should also get this message */
			ROUTER_RESULT("Message result: External relay with CC");
		}
		send_msg_via_ha(xml_relay_message, host_to);
		processing_complete = TRUE & dont_cc;
		
	}
	
	FNRET(processing_complete);
}

void
send_msg_via_ha(xmlNodePtr action, const char *dest_node)
{
	FNIN();
	if (action == NULL) FNOUT();

	if (validate_crm_message(action, NULL, NULL, NULL) == NULL)
	{
		cl_log(LOG_ERR,
		       "Relay message to (%s) via HA was invalid, ignoring",
		       dest_node);
		FNOUT();
	}
/* 	CRM_DEBUG2("Relaying message to (%s) via HA", dest_node); */
	set_xml_property_copy(action, XML_ATTR_HOSTTO, dest_node);

	send_xmlha_message(fsa_cluster_conn, action);
	FNOUT();
}


void
send_msg_via_ipc(xmlNodePtr action, const char *sys)
{
	IPC_Channel *client_channel = NULL;

	FNIN();
/* 	cl_log(LOG_DEBUG, "relaying msg to sub_sys=%s via IPC", sys); */

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

	if (client_channel != NULL) {
		cl_log(LOG_DEBUG, "Sending message via channel %s.", sys);
		send_xmlipc_message(client_channel, action);
	} else if(sys != NULL && strcmp(sys, CRM_SYSTEM_CIB) == 0) {
		cl_log(LOG_ERR,
		       "Sub-system (%s) has been incorporated into the CRMd.",
		       sys);
		xml_message_debug(action, "Change the way we handle");
		relay_message(process_cib_message(action, TRUE), TRUE);
		
	} else if(sys != NULL && strcmp(sys, CRM_SYSTEM_LRMD) == 0) {

		do_lrm_invoke(A_LRM_INVOKE, C_IPC_MESSAGE,
			      fsa_state, I_MESSAGE, action);
		
	} else {
		cl_log(LOG_ERR,
		       "Unknown Sub-system (%s)... discarding message.",
		       sys);
	}    
	FNOUT();
}	

gboolean
crmd_authorize_message(xmlNodePtr root_xml_node,
		       IPC_Message *client_msg,
		       crmd_client_t *curr_client)
{
	/*  check the best case first */
	const char *sys_from   = xmlGetProp(root_xml_node,
					    XML_ATTR_SYSFROM);
	char *uuid = NULL;
	char *client_name = NULL;
	char *major_version = NULL;
	char *minor_version = NULL;
	gboolean result;
	gpointer table_key = NULL;
	
	FNIN();

	if (sys_from != NULL) {
		gboolean can_reply = FALSE; /*  no-one has registered with this id */
		const char *filtered_from = sys_from;

		/* The CIB can have two names on the DC */
		if(strcmp(sys_from, CRM_SYSTEM_DCIB) == 0)
			filtered_from = CRM_SYSTEM_CIB;
		
		if (g_hash_table_lookup (ipc_clients, filtered_from) != NULL)
			can_reply = TRUE;  /*  reply can be routed */
		
		
		CRM_DEBUG3("Message reply can%s be routed from %s.",
			   can_reply?"":" not", sys_from);
		FNRET(can_reply);
	}

	/*  otherwise, check if it was a hello message */

	cl_log(LOG_INFO,
	       "received client join msg: %s",
	       (char*)client_msg->msg_body);

	result = process_hello_message(client_msg,
						&uuid,
						&client_name,
						&major_version,
						&minor_version);

	if (result == TRUE) {
		/*  check version */
		int mav = atoi(major_version);
		int miv = atoi(minor_version);
		if (mav < 0 || miv < 0) {
			cl_log(LOG_ERR,
			       "Client version (%d:%d) is not acceptable",
			       mav,
			       miv);
			result = FALSE;
		}
		cl_free(major_version);
		cl_free(minor_version);
	}

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

		struct crm_subsystem_s *the_subsystem = NULL;
		
		if (client_name == NULL)
			CRM_DEBUG("Client had not registered with us yet");
		else 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;
		else if (strcmp(CRM_SYSTEM_CIB, client_name) == 0)
			the_subsystem = cib_subsystem;

		if (the_subsystem != NULL) {
			/*  do we already have one? */
			result = (fsa_input_register & the_subsystem->flag) == 0;
			set_bit_inplace(&fsa_input_register, the_subsystem->flag);

			if(result) {
				the_subsystem->ipc = curr_client->client_channel;
			} /*  else we didnt ask for the client to start */

		} else if(client_name != NULL && uuid != NULL) {
			table_key = (gpointer)
				generate_hash_key(client_name, uuid);
		} else {
			result = FALSE;
			cl_log(LOG_ERR, "Bad client details (client_name=%s, uuid=%s)",
			       client_name, uuid);
		}
		
	}
	
	if(result == TRUE && table_key == NULL)
		table_key = (gpointer)cl_strdup(client_name);

	if (result == TRUE) {
		cl_log(LOG_INFO, "Accepted client %s", (char*)table_key);

		curr_client->table_key = table_key;
		curr_client->sub_sys = cl_strdup(client_name);
		curr_client->uuid = cl_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");
	} else {
		cl_log(LOG_ERR, "Rejected client logon request");
		curr_client->client_channel->ch_status = IPC_DISC_PENDING;
	}
	
	if(uuid != NULL) cl_free(uuid);
	if(client_name != NULL) cl_free(client_name);

	/* hello messages should never be processed further */
	FNRET(FALSE);
}

enum crmd_fsa_input
handle_message(xmlNodePtr stored_msg)
{
	enum crmd_fsa_input next_input = I_NULL;

	const char *sys_to   = get_xml_attr(stored_msg, NULL,
					    XML_ATTR_SYSTO, TRUE);

/* 	const char *sys_from = get_xml_attr(stored_msg, NULL, */
/* 					    XML_ATTR_SYSFROM, TRUE); */

	const char *type     = get_xml_attr(stored_msg, NULL,
					    XML_ATTR_MSGTYPE, TRUE);
	
	const char *op       = get_xml_attr(stored_msg, XML_TAG_OPTIONS,
					    XML_ATTR_OP, TRUE);

/* 	xml_message_debug(stored_msg, "Processing message"); */

	if(type == NULL || op == NULL) {
		cl_log(LOG_ERR, "Ignoring message (type=%s), (op=%s)",
		       type, op);
		xml_message_debug(stored_msg, "Bad message");
		
	} else if(strcmp(type, XML_ATTR_REQUEST) == 0){
		if(strcmp(op, CRM_OPERATION_VOTE) == 0) {
			next_input = I_ELECTION;
				
		} else if(strcmp(op, CRM_OPERATION_HBEAT) == 0) {
			next_input = I_DC_HEARTBEAT;
				
		} else if(strcmp(op, CRM_OPERATION_WELCOME) == 0) {
			next_input = I_WELCOME;
				
		} else if(strcmp(op, CRM_OPERATION_SHUTDOWN_REQ) == 0) {
			next_input = I_CIB_OP;
				
		} else if(strcmp(op, CRM_OPERATION_SHUTDOWN) == 0) {
			next_input = I_TERMINATE;
			
		} else if(strcmp(op, CRM_OPERATION_ANNOUNCE) == 0) {
			next_input = I_NODE_JOIN;
			
		} else if(strcmp(op, CRM_OPERATION_REPLACE) == 0
			|| strcmp(op, CRM_OPERATION_ERASE) == 0) {
			next_input = I_CIB_OP;
			fprintf(router_strm, "Message result: CIB Op\n");

		} else if(AM_I_DC
			  && (strcmp(op, CRM_OPERATION_CREATE) == 0
			      || strcmp(op, CRM_OPERATION_UPDATE) == 0
			      || strcmp(op, CRM_OPERATION_DELETE) == 0)) {
			/* updates should only be performed on the DC */
			next_input = I_CIB_OP;
				
		} else if(strcmp(op, CRM_OPERATION_PING) == 0) {
			/* eventually do some stuff to figure out
			 * if we /are/ ok
			 */
			xmlNodePtr ping =
				createPingAnswerFragment(sys_to, "ok");
				
			xmlNodePtr wrapper = create_reply(stored_msg, ping);
			relay_message(wrapper, TRUE);
			free_xml(wrapper);
				
		} else {
			cl_log(LOG_ERR,
			       "Unexpected request (op=%s) sent to the %s",
			       op, AM_I_DC?"DC":"CRMd");
		}
		
	} else if(strcmp(type, XML_ATTR_RESPONSE) == 0) {

		if(strcmp(op, CRM_OPERATION_WELCOME) == 0) {
			next_input = I_WELCOME_ACK;
				
		} else if(strcmp(op, CRM_OPERATION_VOTE) == 0
			  || strcmp(op, CRM_OPERATION_HBEAT) == 0
			  || strcmp(op, CRM_OPERATION_WELCOME) == 0
			  || strcmp(op, CRM_OPERATION_SHUTDOWN_REQ) == 0
			  || strcmp(op, CRM_OPERATION_SHUTDOWN) == 0
			  || strcmp(op, CRM_OPERATION_ANNOUNCE) == 0) {
			next_input = I_NULL;
			
/* 		} else if(AM_I_DC */
/* 			  && (strcmp(op, CRM_OPERATION_CREATE) == 0 */
/* 			      || strcmp(op, CRM_OPERATION_UPDATE) == 0 */
/* 			      || strcmp(op, CRM_OPERATION_DELETE) == 0 */
/* 			      || strcmp(op, CRM_OPERATION_REPLACE) == 0 */
/* 			      || strcmp(op, CRM_OPERATION_ERASE) == 0)) { */

   			/*  perhaps we should do somethign with these replies */
/* 			fprintf(router_strm, "Message result: CIB Reply\n"); */

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

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


void lrm_op_callback (lrm_op_t* op)
{
	CRM_DEBUG("In lrm_op_callback()");

	s_crmd_fsa(C_LRM_OP_CALLBACK, I_LRM_EVENT, op);
}

void lrm_monitor_callback (lrm_mon_t* monitor)
{
	CRM_DEBUG("In lrm_monitor_callback()");
	s_crmd_fsa(C_LRM_MONITOR_CALLBACK, I_LRM_EVENT, monitor);
}