diff --git a/crm/crmd/callbacks.c b/crm/crmd/callbacks.c index 757c10ff56..f9e5af2927 100644 --- a/crm/crmd/callbacks.c +++ b/crm/crmd/callbacks.c @@ -1,618 +1,614 @@ /* * Copyright (C) 2004 Andrew Beekhof * * 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.1 of the License, or (at your option) any later version. * * This software 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 library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include #include #include #include #include #include #include #include #include GHashTable *crmd_peer_state = NULL; crm_data_t *find_xml_in_hamessage(const HA_Message * msg); void crmd_ha_connection_destroy(gpointer user_data); /* From join_dc... */ extern gboolean check_join_state( enum crmd_fsa_state cur_state, const char *source); /* #define MAX_EMPTY_CALLBACKS 20 */ /* int empty_callbacks = 0; */ gboolean crmd_ha_msg_dispatch(IPC_Channel *channel, gpointer user_data) { int lpc = 0; ll_cluster_t *hb_cluster = (ll_cluster_t*)user_data; while(lpc < 2 && hb_cluster->llc_ops->msgready(hb_cluster)) { if(channel->ch_status != IPC_CONNECT) { /* there really is no point continuing */ break; } lpc++; /* invoke the callbacks but dont block */ hb_cluster->llc_ops->rcvmsg(hb_cluster, 0); } crm_devel("%d HA messages dispatched", lpc); G_main_set_trigger(fsa_source); if (channel && (channel->ch_status != IPC_CONNECT)) { crm_crit("Lost connection to heartbeat service."); return FALSE; } return TRUE; } void crmd_ha_msg_callback(const HA_Message * msg, void* private_data) { ha_msg_input_t *new_input = NULL; oc_node_t *from_node = NULL; const char *from = ha_msg_value(msg, F_ORIG); const char *seq = ha_msg_value(msg, F_SEQ); const char *op = ha_msg_value(msg, F_CRM_TASK); const char *sys_to = ha_msg_value(msg, F_CRM_SYS_TO); const char *sys_from = ha_msg_value(msg, F_CRM_SYS_FROM); CRM_DEV_ASSERT(from != NULL); if(fsa_membership_copy == NULL) { crm_debug("Ignoring HA messages until we are" " connected to the CCM (%s op from %s)", op, from); crm_log_message_adv( LOG_MSG, "HA[inbound]: Ignore (No CCM)", msg); return; } from_node = g_hash_table_lookup(fsa_membership_copy->members, from); if(from_node == NULL) { int level = LOG_DEBUG; if(safe_str_eq(op, CRM_OP_VOTE)) { level = LOG_WARNING; } else if(AM_I_DC && safe_str_eq(op, CRM_OP_JOIN_ANNOUNCE)) { level = LOG_WARNING; } else if(safe_str_eq(sys_from, CRM_SYSTEM_DC)) { level = LOG_WARNING; } do_crm_log(level, __FILE__, __FUNCTION__, "Ignoring HA message (op=%s) from %s: not in our" " membership list (size=%d)", op, from, g_hash_table_size(fsa_membership_copy->members)); crm_log_message_adv(LOG_MSG, "HA[inbound]: CCM Discard", msg); } else if(AM_I_DC && safe_str_eq(sys_from, CRM_SYSTEM_DC) && safe_str_neq(from, fsa_our_uname)) { crm_err("Another DC detected: %s (op=%s)", from, op); crm_log_message_adv( LOG_WARNING, "HA[inbound]: Duplicate DC", msg); new_input = new_ha_msg_input(msg); /* make sure the election happens NOW */ register_fsa_error_adv(C_FSA_INTERNAL, I_ELECTION, NULL, new_input, __FUNCTION__); #if 0 /* still thinking about this one... * could create a timing issue if we dont notice the * election before a new DC is elected. */ } else if(fsa_our_dc != NULL && safe_str_eq(sys_from, CRM_SYSTEM_DC) && safe_str_neq(from, fsa_our_dc)) { crm_warn("Ignoring message from wrong DC: %s vs. %s ", from, fsa_our_dc); crm_log_message_adv(LOG_WARNING, "HA[inbound]: wrong DC", msg); #endif } else if(safe_str_eq(sys_to, CRM_SYSTEM_DC) && AM_I_DC == FALSE) { crm_verbose("Ignoring message for the DC [F_SEQ=%s]", seq); crm_log_message_adv(LOG_TRACE, "HA[inbound]: ignore", msg); return; } else if(safe_str_eq(from, fsa_our_uname) && safe_str_eq(op, CRM_OP_VOTE)) { crm_log_message_adv(LOG_TRACE, "HA[inbound]", msg); crm_verbose("Ignoring our own vote [F_SEQ=%s]: own vote", seq); return; } else if(AM_I_DC && safe_str_eq(op, CRM_OP_HBEAT)) { crm_verbose("Ignoring our own heartbeat [F_SEQ=%s]", seq); crm_log_message_adv(LOG_TRACE, "HA[inbound]: own heartbeat", msg); return; } else { crm_devel("Processing message"); crm_log_message_adv(LOG_MSG, "HA[inbound]", msg); new_input = new_ha_msg_input(msg); register_fsa_input(C_HA_MESSAGE, I_ROUTER, new_input); } #if 0 if(ha_msg_value(msg, XML_ATTR_REFERENCE) == NULL) { ha_msg_add(new_input->msg, XML_ATTR_REFERENCE, seq); } #endif delete_ha_msg_input(new_input); return; } /* * Apparently returning TRUE means "stay connected, keep doing stuff". * Returning FALSE means "we're all done, close the connection" */ gboolean crmd_ipc_msg_callback(IPC_Channel *client, gpointer user_data) { int lpc = 0; IPC_Message *msg = NULL; ha_msg_input_t *new_input = NULL; crmd_client_t *curr_client = (crmd_client_t*)user_data; gboolean stay_connected = TRUE; crm_verbose("Processing IPC message from %s", curr_client->table_key); while(lpc == 0 && client->ops->is_message_pending(client)) { if (client->ch_status != IPC_CONNECT) { /* The message which was pending for us is that * the IPC status is now IPC_DISCONNECT */ break; } if (client->ops->recv(client, &msg) != IPC_OK) { perror("Receive failure:"); crm_err("[%s] [receive failure]", curr_client->table_key); stay_connected = FALSE; break; } else if (msg == NULL) { crm_err("No message from %s this time", curr_client->table_key); continue; } lpc++; new_input = new_ipc_msg_input(msg); msg->msg_done(msg); crm_verbose("Processing msg from %s", curr_client->table_key); crm_log_message_adv(LOG_MSG, "CRMd[inbound]", new_input->msg); crmd_authorize_message(new_input, curr_client); delete_ha_msg_input(new_input); msg = NULL; new_input = NULL; } crm_verbose("Processed %d messages", lpc); if (client->ch_status != IPC_CONNECT) { stay_connected = FALSE; crm_verbose("received HUP from %s", curr_client->table_key); if (curr_client != NULL) { struct crm_subsystem_s *the_subsystem = NULL; if (curr_client->sub_sys == NULL) { crm_debug("Client hadn't registered with us yet"); } else if (strcmp(CRM_SYSTEM_PENGINE, curr_client->sub_sys) == 0) { the_subsystem = pe_subsystem; } else if (strcmp(CRM_SYSTEM_TENGINE, curr_client->sub_sys) == 0) { the_subsystem = te_subsystem; } else if (strcmp(CRM_SYSTEM_CIB, curr_client->sub_sys) == 0){ the_subsystem = cib_subsystem; } if(the_subsystem != NULL) { cleanup_subsystem(the_subsystem); } /* else that was a transient client */ if (curr_client->table_key != NULL) { /* * Key is destroyed below: * curr_client->table_key * Value is cleaned up by: * G_main_del_IPC_Channel */ g_hash_table_remove( ipc_clients, curr_client->table_key); } #if 0 if(curr_client->client_source != NULL) { gboolean det = G_main_del_IPC_Channel( curr_client->client_source); crm_verbose("crm_client was %s detached", det?"successfully":"not"); } #endif crm_free(curr_client->table_key); crm_free(curr_client->sub_sys); crm_free(curr_client->uuid); crm_free(curr_client); } } G_main_set_trigger(fsa_source); return stay_connected; } gboolean lrm_dispatch(IPC_Channel*src_not_used, gpointer user_data) { int num_msgs = 0; ll_lrm_t *lrm = (ll_lrm_t*)user_data; crm_devel("received callback"); num_msgs = lrm->lrm_ops->rcvmsg(lrm, FALSE); if(num_msgs < 1) { crm_err("lrm->lrm_ops->rcvmsg() failed, connection lost?"); clear_bit_inplace(fsa_input_register, R_LRM_CONNECTED); register_fsa_input(C_FSA_INTERNAL, I_ERROR, NULL); - G_main_set_trigger(fsa_source); return FALSE; } return TRUE; } void lrm_op_callback(lrm_op_t* op) { CRM_DEV_ASSERT(op != NULL); if(crm_assert_failed) { return; } crm_debug("received callback: %s/%s (%s)", op->op_type, op->rsc_id, op_status2text(op->op_status)); /* Make sure the LRM events are received in order */ register_fsa_input_later(C_LRM_OP_CALLBACK, I_LRM_EVENT, op); - G_main_set_trigger(fsa_source); } void crmd_ha_status_callback( const char *node, const char * status, void* private_data) { crm_data_t *update = NULL; crm_devel("received callback"); crm_notice("Status update: Node %s now has status [%s]",node,status); if(safe_str_neq(status, DEADSTATUS)) { crm_devel("nstatus callback was not for a dead node"); return; } /* this node is taost */ update = create_node_state( node, node, status, NULL, NULL, NULL, NULL); set_xml_property_copy( update, XML_CIB_ATTR_CLEAR_SHUTDOWN, XML_BOOLEAN_TRUE); update_local_cib(create_cib_fragment(update, NULL)); G_main_set_trigger(fsa_source); free_xml(update); } void crmd_client_status_callback(const char * node, const char * client, const char * status, void * private) { const char *join = NULL; const char *extra = NULL; crm_data_t * update = NULL; crm_devel("received callback"); if(safe_str_neq(client, CRM_SYSTEM_CRMD)) { return; } if(safe_str_eq(status, JOINSTATUS)){ status = ONLINESTATUS; extra = XML_CIB_ATTR_CLEAR_SHUTDOWN; } else if(safe_str_eq(status, LEAVESTATUS)){ status = OFFLINESTATUS; join = CRMD_JOINSTATE_DOWN; extra = XML_CIB_ATTR_CLEAR_SHUTDOWN; } set_bit_inplace(fsa_input_register, R_PEER_DATA); g_hash_table_replace( crmd_peer_state, crm_strdup(node), crm_strdup(status)); if(fsa_state == S_STARTING || fsa_state == S_STOPPING) { return; } crm_notice("Status update: Client %s/%s now has status [%s]", node, client, status); if(safe_str_eq(node, fsa_our_dc) && safe_str_eq(status, OFFLINESTATUS)) { /* did our DC leave us */ crm_info("Got client status callback - our DC is dead"); register_fsa_input(C_CRMD_STATUS_CALLBACK, I_ELECTION, NULL); } else { crm_devel("Got client status callback"); update = create_node_state( node, node, NULL, NULL, status, join, NULL); set_xml_property_copy(update, extra, XML_BOOLEAN_TRUE); update_local_cib(create_cib_fragment(update, NULL)); free_xml(update); if(AM_I_DC && safe_str_eq(status, OFFLINESTATUS)) { g_hash_table_remove(confirmed_nodes, node); g_hash_table_remove(finalized_nodes, node); g_hash_table_remove(integrated_nodes, node); g_hash_table_remove(welcomed_nodes, node); check_join_state(fsa_state, __FUNCTION__); } } G_main_set_trigger(fsa_source); } void crmd_ha_connection_destroy(gpointer user_data) { crm_crit("Heartbeat has left us"); /* this is always an error */ /* feed this back into the FSA */ register_fsa_input(C_HA_DISCONNECT, I_ERROR, NULL); - G_main_set_trigger(fsa_source); } gboolean crmd_client_connect(IPC_Channel *client_channel, gpointer user_data) { if (client_channel == NULL) { crm_err("Channel was NULL"); } else if (client_channel->ch_status == IPC_DISCONNECT) { crm_err("Channel was disconnected"); } else { crmd_client_t *blank_client = NULL; crm_devel("Channel connected"); crm_malloc0(blank_client, sizeof(crmd_client_t)); if (blank_client == NULL) { return FALSE; } client_channel->ops->set_recv_qlen(client_channel, 100); client_channel->ops->set_send_qlen(client_channel, 100); blank_client->client_channel = client_channel; blank_client->sub_sys = NULL; blank_client->uuid = NULL; blank_client->table_key = NULL; blank_client->client_source = G_main_add_IPC_Channel( G_PRIORITY_LOW, client_channel, FALSE, crmd_ipc_msg_callback, blank_client, default_ipc_connection_destroy); } return TRUE; } gboolean ccm_dispatch(int fd, gpointer user_data) { int rc = 0; oc_ev_t *ccm_token = (oc_ev_t*)user_data; gboolean was_error = FALSE; crm_devel("received callback"); rc = oc_ev_handle_event(ccm_token); if(rc != 0) { crm_err("CCM connection appears to have failed: rc=%d.", rc); register_fsa_input(C_CCM_CALLBACK, I_ERROR, NULL); was_error = TRUE; } G_main_set_trigger(fsa_source); return !was_error; } static gboolean fsa_have_quorum = FALSE; void crmd_ccm_msg_callback( oc_ed_t event, void *cookie, size_t size, const void *data) { int instance = -1; gboolean update_cache = FALSE; struct crmd_ccm_data_s *event_data = NULL; const oc_ev_membership_t *membership = data; gboolean update_quorum = FALSE; gboolean trigger_transition = FALSE; crm_devel("received callback"); if(data != NULL) { instance = membership->m_instance; } crm_info("Quorum %s after event=%s (id=%d)", ccm_have_quorum(event)?"(re)attained":"lost", ccm_event_name(event), instance); switch(event) { case OC_EV_MS_NEW_MEMBERSHIP: case OC_EV_MS_INVALID: update_cache = TRUE; update_quorum = TRUE; if(AM_I_DC && need_transition(fsa_state)) { trigger_transition = TRUE; } break; case OC_EV_MS_NOT_PRIMARY: #if UNTESTED if(AM_I_DC == FALSE) { break; } /* tell the TE to pretend it completed and stop */ /* side effect: we'll end up in S_IDLE */ register_fsa_action(A_TE_HALT, TRUE); #endif break; case OC_EV_MS_PRIMARY_RESTORED: if(AM_I_DC && need_transition(fsa_state)) { fsa_membership_copy->id = instance; trigger_transition = TRUE; } break; case OC_EV_MS_EVICTED: update_quorum = TRUE; register_fsa_input(C_FSA_INTERNAL, I_STOP, NULL); break; default: crm_err("Unknown CCM event: %d", event); } if(update_quorum && ccm_have_quorum(event) == FALSE) { /* did we just loose quorum? */ if(fsa_have_quorum && need_transition(fsa_state)) { crm_info("Quorum lost: triggering transition (%s)", ccm_event_name(event)); trigger_transition = TRUE; } fsa_have_quorum = FALSE; } else if(update_quorum) { crm_debug("Updating quorum after event %s", ccm_event_name(event)); fsa_have_quorum = TRUE; } if(update_cache) { crm_debug("Updating cache after event %s", ccm_event_name(event)); crm_malloc0(event_data, sizeof(struct crmd_ccm_data_s)); if(event_data == NULL) { return; } event_data->event = event; if(data != NULL) { event_data->oc = copy_ccm_oc_data(data); } register_fsa_input_adv( C_CCM_CALLBACK, I_CCM_EVENT, event_data, trigger_transition?A_TE_CANCEL:A_NOTHING, FALSE, __FUNCTION__); if (event_data->oc) { crm_free(event_data->oc); event_data->oc = NULL; } crm_free(event_data); } else if(trigger_transition) { crm_debug("Scheduling transition after event %s", ccm_event_name(event)); register_fsa_action(A_TE_CANCEL); } oc_ev_callback_done(cookie); return; } void crmd_cib_connection_destroy(gpointer user_data) { if(is_set(fsa_input_register, R_SHUTDOWN)) { crm_info("Connection to the CIB terminated..."); return; } /* eventually this will trigger a reconnect, not a shutdown */ crm_err("Connection to the CIB terminated..."); register_fsa_input(C_FSA_INTERNAL, I_ERROR, NULL); clear_bit_inplace(fsa_input_register, R_CIB_CONNECTED); - G_main_set_trigger(fsa_source); return; } longclock_t fsa_start = 0; longclock_t fsa_stop = 0; longclock_t fsa_diff = 0; gboolean crm_fsa_trigger(gpointer user_data) { unsigned int fsa_diff_ms = 0; if(fsa_diff_max_ms > 0) { fsa_start = time_longclock(); } s_crmd_fsa(C_FSA_INTERNAL); if(fsa_diff_max_ms > 0) { fsa_stop = time_longclock(); fsa_diff = sub_longclock(fsa_stop, fsa_start); fsa_diff_ms = longclockto_ms(fsa_diff); if(fsa_diff_ms < 0 || fsa_diff_ms > fsa_diff_max_ms) { crm_err("FSA took %dms to complete", fsa_diff_ms); } } return TRUE; } diff --git a/crm/crmd/control.c b/crm/crmd/control.c index 9930da14e4..07b23383ad 100644 --- a/crm/crmd/control.c +++ b/crm/crmd/control.c @@ -1,666 +1,664 @@ /* * Copyright (C) 2004 Andrew Beekhof * * 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.1 of the License, or (at your option) any later version. * * This software 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 library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include #include #include #include #include #include #include #include extern void crmd_ha_connection_destroy(gpointer user_data); extern gboolean stop_all_resources(void); gboolean crm_shutdown(int nsig, gpointer unused); gboolean register_with_ha(ll_cluster_t *hb_cluster, const char *client_name); GHashTable *ipc_clients = NULL; GTRIGSource *fsa_source = NULL; /* A_HA_CONNECT */ enum crmd_fsa_input do_ha_control(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { gboolean registered = FALSE; if(action & A_HA_DISCONNECT) { if(fsa_cluster_conn != NULL) { fsa_cluster_conn->llc_ops->signoff( fsa_cluster_conn, FALSE); } } if(action & A_HA_CONNECT) { if(fsa_cluster_conn == NULL) { fsa_cluster_conn = ll_cluster_new("heartbeat"); } /* make sure we are disconnected first */ fsa_cluster_conn->llc_ops->signoff(fsa_cluster_conn, FALSE); registered = register_with_ha( fsa_cluster_conn, crm_system_name); if(registered == FALSE) { register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL); return I_NULL; } } if(action & ~(A_HA_CONNECT|A_HA_DISCONNECT)) { crm_err("Unexpected action %s in %s", fsa_action2string(action), __FUNCTION__); } return I_NULL; } /* A_SHUTDOWN */ enum crmd_fsa_input do_shutdown(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { enum crmd_fsa_input next_input = I_NULL; enum crmd_fsa_input tmp = I_NULL; /* just in case */ set_bit_inplace(fsa_input_register, R_SHUTDOWN); /* last attempt to shut these down */ if(is_set(fsa_input_register, R_PE_CONNECTED)) { crm_warn("Last attempt to shutdown the PolicyEngine"); tmp = do_pe_control(A_PE_STOP, cause, cur_state, current_input, msg_data); if(tmp != I_NULL) { next_input = I_ERROR; crm_err("Failed to shutdown the PolicyEngine"); } } if(is_set(fsa_input_register, R_TE_CONNECTED)) { crm_warn("Last attempt to shutdown the Transitioner"); tmp = do_pe_control(A_TE_STOP, cause, cur_state, current_input, msg_data); if(tmp != I_NULL) { next_input = I_ERROR; crm_err("Failed to shutdown the Transitioner"); } } crm_info("Stopping all remaining local resources"); if(is_set(fsa_input_register, R_LRM_CONNECTED)) { stop_all_resources(); } else { crm_err("Exiting with no LRM connection..." " resources may be active!"); register_fsa_input(C_FSA_INTERNAL, I_TERMINATE, NULL); } return next_input; } /* A_SHUTDOWN_REQ */ enum crmd_fsa_input do_shutdown_req(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { enum crmd_fsa_input next_input = I_NULL; HA_Message *msg = NULL; crm_info("Sending shutdown request to DC: %s", crm_str(fsa_our_dc)); msg = create_request( CRM_OP_SHUTDOWN_REQ, NULL, NULL, CRM_SYSTEM_DC, CRM_SYSTEM_CRMD, NULL); /* set_bit_inplace(fsa_input_register, R_STAYDOWN); */ if(send_request(msg, NULL) == FALSE) { next_input = I_ERROR; } else { crm_timer_start(shutdown_timer); } return next_input; } /* A_EXIT_0, A_EXIT_1 */ enum crmd_fsa_input do_exit(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { int exit_code = 0; gboolean do_exit = FALSE; if(action & A_EXIT_0) { if(is_set(fsa_input_register, R_PE_CONNECTED)) { crm_info("Waiting for the PE to disconnect"); crmd_fsa_stall(); } else if(is_set(fsa_input_register, R_TE_CONNECTED)) { crm_info("Waiting for the TE to disconnect"); crmd_fsa_stall(); } else { do_exit = TRUE; crm_info("Performing %s - gracefully exiting the CRMd", fsa_action2string(action)); } } else { do_exit = TRUE; exit_code = 1; crm_warn("Performing %s - forcefully exiting the CRMd... now!", fsa_action2string(action)); } if(do_exit) { if(is_set(fsa_input_register, R_IN_RECOVERY)) { crm_info("Could not recover from internal error"); exit_code = 2; } else if(is_set(fsa_input_register, R_STAYDOWN)) { crm_info("Inhibiting respawn by Heartbeat"); exit_code = 100; } crm_info("[%s] stopped (%d)", crm_system_name, exit_code); exit(exit_code); } return I_NULL; } /* A_STARTUP */ enum crmd_fsa_input do_startup(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { int was_error = 0; int interval = 1; /* seconds between DC heartbeats */ crm_info("Register Signal Handler"); G_main_add_SignalHandler( G_PRIORITY_HIGH, SIGTERM, crm_shutdown, NULL, NULL); fsa_source = G_main_add_TriggerHandler( G_PRIORITY_HIGH, crm_fsa_trigger, NULL, NULL); ipc_clients = g_hash_table_new(g_str_hash, g_str_equal); if(was_error == 0) { crm_info("Init server comms"); was_error = init_server_ipc_comms( crm_strdup(CRM_SYSTEM_CRMD), crmd_client_connect, default_ipc_connection_destroy); } if(was_error == 0) { crm_info("Creating CIB object"); fsa_cib_conn = cib_new(); } /* set up the timers */ crm_malloc0(dc_heartbeat, sizeof(fsa_timer_t)); crm_malloc0(integration_timer, sizeof(fsa_timer_t)); crm_malloc0(finalization_timer, sizeof(fsa_timer_t)); crm_malloc0(election_trigger, sizeof(fsa_timer_t)); crm_malloc0(election_timeout, sizeof(fsa_timer_t)); crm_malloc0(shutdown_escalation_timer, sizeof(fsa_timer_t)); crm_malloc0(wait_timer, sizeof(fsa_timer_t)); crm_malloc0(shutdown_timer, sizeof(fsa_timer_t)); interval = interval * 1000; if(election_trigger != NULL) { election_trigger->source_id = -1; election_trigger->period_ms = -1; election_trigger->fsa_input = I_DC_TIMEOUT; election_trigger->callback = crm_timer_popped; election_trigger->repeat = FALSE; } else { was_error = TRUE; } if(dc_heartbeat != NULL) { dc_heartbeat->source_id = -1; dc_heartbeat->period_ms = -1; dc_heartbeat->fsa_input = I_NULL; dc_heartbeat->callback = do_dc_heartbeat; dc_heartbeat->repeat = FALSE; } else { was_error = TRUE; } if(election_timeout != NULL) { election_timeout->source_id = -1; election_timeout->period_ms = -1; election_timeout->fsa_input = I_ELECTION_DC; election_timeout->callback = crm_timer_popped; election_timeout->repeat = FALSE; } else { was_error = TRUE; } if(integration_timer != NULL) { integration_timer->source_id = -1; integration_timer->period_ms = -1; integration_timer->fsa_input = I_INTEGRATED; integration_timer->callback = crm_timer_popped; integration_timer->repeat = FALSE; } else { was_error = TRUE; } if(finalization_timer != NULL) { finalization_timer->source_id = -1; finalization_timer->period_ms = -1; finalization_timer->fsa_input = I_FINALIZED; finalization_timer->callback = crm_timer_popped; finalization_timer->repeat = FALSE; } else { was_error = TRUE; } if(shutdown_escalation_timer != NULL) { shutdown_escalation_timer->source_id = -1; shutdown_escalation_timer->period_ms = -1; shutdown_escalation_timer->fsa_input = I_STOP; shutdown_escalation_timer->callback = crm_timer_popped; shutdown_escalation_timer->repeat = FALSE; } else { was_error = TRUE; } if(wait_timer != NULL) { wait_timer->source_id = -1; wait_timer->period_ms = 500; wait_timer->fsa_input = I_NULL; wait_timer->callback = crm_timer_popped; wait_timer->repeat = FALSE; } else { was_error = TRUE; } if(shutdown_timer != NULL) { shutdown_timer->source_id = -1; shutdown_timer->period_ms = -1; shutdown_timer->fsa_input = I_SHUTDOWN; shutdown_timer->callback = crm_timer_popped; shutdown_timer->repeat = TRUE; } else { was_error = TRUE; } /* set up the sub systems */ crm_malloc0(cib_subsystem, sizeof(struct crm_subsystem_s)); crm_malloc0(te_subsystem, sizeof(struct crm_subsystem_s)); crm_malloc0(pe_subsystem, sizeof(struct crm_subsystem_s)); if(cib_subsystem != NULL) { cib_subsystem->pid = -1; cib_subsystem->path = BIN_DIR; cib_subsystem->name = CRM_SYSTEM_CIB; cib_subsystem->command = BIN_DIR"/"CRM_SYSTEM_CIB; cib_subsystem->args = "-VVc"; cib_subsystem->flag_connected = R_CIB_CONNECTED; cib_subsystem->flag_required = R_CIB_REQUIRED; } else { was_error = TRUE; } if(te_subsystem != NULL) { te_subsystem->pid = -1; te_subsystem->path = BIN_DIR; te_subsystem->name = CRM_SYSTEM_TENGINE; te_subsystem->command = BIN_DIR"/"CRM_SYSTEM_TENGINE; te_subsystem->args = NULL; te_subsystem->flag_connected = R_TE_CONNECTED; te_subsystem->flag_required = R_TE_REQUIRED; } else { was_error = TRUE; } if(pe_subsystem != NULL) { pe_subsystem->pid = -1; pe_subsystem->path = BIN_DIR; pe_subsystem->name = CRM_SYSTEM_PENGINE; pe_subsystem->command = BIN_DIR"/"CRM_SYSTEM_PENGINE; pe_subsystem->args = NULL; pe_subsystem->flag_connected = R_PE_CONNECTED; pe_subsystem->flag_required = R_PE_REQUIRED; } else { was_error = TRUE; } if(was_error) { register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL); } welcomed_nodes = g_hash_table_new_full( g_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str); integrated_nodes = g_hash_table_new_full( g_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str); finalized_nodes = g_hash_table_new_full( g_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str); confirmed_nodes = g_hash_table_new_full( g_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str); crmd_peer_state = g_hash_table_new_full( g_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str); return I_NULL; } extern GHashTable *shutdown_ops; /* A_STOP */ enum crmd_fsa_input do_stop(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { if(g_hash_table_size(shutdown_ops) > 0) { crm_err("%d stop operations outstanding at exit", g_hash_table_size(shutdown_ops)); } return I_NULL; } /* A_STARTED */ enum crmd_fsa_input do_started(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { if(is_set(fsa_input_register, R_CCM_DATA) == FALSE) { crm_info("Delaying start, CCM (%.16llx) not connected", R_CCM_DATA); crmd_fsa_stall(); return I_NULL; } else if(is_set(fsa_input_register, R_LRM_CONNECTED) == FALSE) { crm_info("Delaying start, LRM (%.16llx) not connected", R_LRM_CONNECTED); crmd_fsa_stall(); return I_NULL; } else if(is_set(fsa_input_register, R_CIB_CONNECTED) == FALSE) { crm_info("Delaying start, CIB (%.16llx) not connected", R_CIB_CONNECTED); crmd_fsa_stall(); return I_NULL; } else if(is_set(fsa_input_register, R_PEER_DATA) == FALSE) { HA_Message * msg = NULL; /* try reading from HA */ crm_info("Delaying start, Peer data (%.16llx) not recieved", R_PEER_DATA); crm_devel("Looking for a HA message"); msg = fsa_cluster_conn->llc_ops->readmsg(fsa_cluster_conn, 0); if(msg != NULL) { crm_devel("There was a HA message"); crm_msg_del(msg); } crm_timer_start(wait_timer); crmd_fsa_stall(); return I_NULL; } crm_info("The local CRM is operational"); clear_bit_inplace(fsa_input_register, R_STARTING); register_fsa_input(msg_data->fsa_cause, I_PENDING, NULL); return I_NULL; } /* A_RECOVER */ enum crmd_fsa_input do_recover(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { set_bit_inplace(fsa_input_register, R_IN_RECOVERY); crm_err("Action %s (%.16llx) not supported", fsa_action2string(action), action); register_fsa_input(C_FSA_INTERNAL, I_STOP, NULL); return I_NULL; } /* A_READCONFIG */ enum crmd_fsa_input do_read_config(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { /* this one probably is worthwhile blocking on */ crm_data_t *cib_copy = get_cib_copy(fsa_cib_conn); crm_data_t *config = get_object_root(XML_CIB_TAG_CRMCONFIG, cib_copy); dc_heartbeat->period_ms = 0; xml_child_iter( config, iter, XML_CIB_TAG_NVPAIR, const char *name = crm_element_value(iter, XML_NVPAIR_ATTR_NAME); const char *value = crm_element_value(iter, XML_NVPAIR_ATTR_VALUE); if(name == NULL || value == NULL) { continue; } else if(safe_str_eq(name, XML_CONFIG_ATTR_DC_BEAT)) { dc_heartbeat->period_ms = atoi(value); } else if(safe_str_eq(name, XML_CONFIG_ATTR_DC_DEADTIME)) { election_trigger->period_ms = atoi(value); } else if(safe_str_eq(name, XML_CONFIG_ATTR_FORCE_QUIT)) { shutdown_escalation_timer->period_ms = atoi(value); } ); if(dc_heartbeat->period_ms < 1) { /* sensible default */ dc_heartbeat->period_ms = crm_get_msec( getenv("HA_"KEY_KEEPALIVE)); } election_timeout->period_ms = dc_heartbeat->period_ms * 6; integration_timer->period_ms = dc_heartbeat->period_ms * 6; finalization_timer->period_ms = dc_heartbeat->period_ms * 6; integration_timer->period_ms = crm_get_msec("5min"); finalization_timer->period_ms = crm_get_msec("5min"); if(election_trigger->period_ms < 1 || election_trigger->period_ms > election_timeout->period_ms) { /* sensible default */ election_trigger->period_ms = election_timeout->period_ms * 2; } if(shutdown_escalation_timer->period_ms < 1 || election_timeout->period_ms > shutdown_escalation_timer->period_ms) { /* sensible default - 32 election cycles */ shutdown_escalation_timer->period_ms = (election_timeout->period_ms + election_trigger->period_ms) * 32; } shutdown_timer->period_ms = election_trigger->period_ms; return I_NULL; } gboolean crm_shutdown(int nsig, gpointer unused) { if (crmd_mainloop != NULL && g_main_is_running(crmd_mainloop)) { if(is_set(fsa_input_register, R_SHUTDOWN)) { crm_err("Escalating the shutdown"); register_fsa_input_before(C_SHUTDOWN, I_ERROR, NULL); - G_main_set_trigger(fsa_source); } else { set_bit_inplace(fsa_input_register, R_SHUTDOWN); /* set_bit_inplace(fsa_input_register, R_STAYDOWN); */ /* if we ever win an election we're the last man standing */ election_timeout->fsa_input = I_STOP; if(is_set(fsa_input_register, R_SHUTDOWN)) { /* cant rely on this... */ crm_timer_start(shutdown_escalation_timer); - register_fsa_input(C_SHUTDOWN, I_SHUTDOWN, NULL); - G_main_set_trigger(fsa_source); + register_fsa_input(C_SHUTDOWN,I_SHUTDOWN,NULL); } else { crm_err("Could not set R_SHUTDOWN"); exit(LSB_EXIT_ENOTSUPPORTED); } } } else { crm_info("exit from shutdown"); exit(LSB_EXIT_OK); } return TRUE; } gboolean register_with_ha(ll_cluster_t *hb_cluster, const char *client_name) { if(safe_val3(NULL, hb_cluster, llc_ops, errmsg) == NULL) { crm_crit("cluster errmsg function unavailable"); } crm_debug("Signing in with Heartbeat"); if (hb_cluster->llc_ops->signon(hb_cluster, client_name)!= HA_OK) { crm_err("Cannot sign on with heartbeat: %s", hb_cluster->llc_ops->errmsg(hb_cluster)); return FALSE; } crm_set_ha_options(hb_cluster); crm_devel("Be informed of CRM messages"); if (HA_OK != hb_cluster->llc_ops->set_msg_callback( hb_cluster, T_CRM, crmd_ha_msg_callback, hb_cluster)){ crm_err("Cannot set msg callback: %s", hb_cluster->llc_ops->errmsg(hb_cluster)); return FALSE; } #if 0 crm_devel("Be informed of Node Status changes"); if (HA_OK != hb_cluster->llc_ops->set_nstatus_callback( hb_cluster, crmd_ha_status_callback, hb_cluster)){ crm_err("Cannot set nstatus callback: %s", hb_cluster->llc_ops->errmsg(hb_cluster)); return FALSE; } #endif crm_devel("Be informed of CRM Client Status changes"); if (HA_OK != hb_cluster->llc_ops->set_cstatus_callback( hb_cluster, crmd_client_status_callback, hb_cluster)) { crm_err("Cannot set cstatus callback: %s", hb_cluster->llc_ops->errmsg(hb_cluster)); return FALSE; } crm_devel("Adding channel to mainloop"); G_main_add_IPC_Channel( G_PRIORITY_HIGH, hb_cluster->llc_ops->ipcchan(hb_cluster), FALSE, crmd_ha_msg_dispatch, hb_cluster /* userdata */, crmd_ha_connection_destroy); crm_devel("Finding our node name"); if ((fsa_our_uname = hb_cluster->llc_ops->get_mynodeid(hb_cluster)) == NULL) { crm_err("get_mynodeid() failed"); return FALSE; } crm_info("FSA Hostname: %s", fsa_our_uname); /* Async get client status information in the cluster */ crm_devel("Requesting an initial dump of CRMD client_status"); fsa_cluster_conn->llc_ops->client_status( fsa_cluster_conn, NULL, CRM_SYSTEM_CRMD, -1); return TRUE; } diff --git a/crm/crmd/crmd_messages.h b/crm/crmd/crmd_messages.h index 6029b7d4ef..784fcfd1cb 100644 --- a/crm/crmd/crmd_messages.h +++ b/crm/crmd/crmd_messages.h @@ -1,108 +1,111 @@ /* * Copyright (C) 2004 Andrew Beekhof * * 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.1 of the License, or (at your option) any later version. * * This software 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 library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #ifndef XML_CRM_MESSAGES__H #define XML_CRM_MESSAGES__H #include #include #include #include extern void *fsa_typed_data_adv( fsa_data_t *fsa_data, enum fsa_data_type a_type, const char *caller); #define fsa_typed_data(x) fsa_typed_data_adv(msg_data, x, __FUNCTION__) extern void register_fsa_error_adv( enum crmd_fsa_cause cause, enum crmd_fsa_input input, fsa_data_t *cur_data, void *new_data, const char *raised_from); #define register_fsa_error(cause, input, new_data) register_fsa_error_adv(cause, input, msg_data, new_data, __FUNCTION__) extern void register_fsa_input_adv( enum crmd_fsa_cause cause, enum crmd_fsa_input input, void *data, long long with_actions, gboolean prepend, const char *raised_from); extern void fsa_dump_queue(int log_level); #define crmd_fsa_stall() register_fsa_input_adv(msg_data->fsa_cause, I_WAIT_FOR_EVENT, msg_data->data, action, TRUE, __FUNCTION__) #define register_fsa_input(cause, input, data) register_fsa_input_adv(cause, input, data, A_NOTHING, FALSE, __FUNCTION__) #define register_fsa_action(action) { \ fsa_actions |= action; \ + if(fsa_source) { \ + G_main_set_trigger(fsa_source); \ + } \ crm_debug("%s added action %s to the FSA", \ __FUNCTION__, fsa_action2string(action)); \ } #define register_fsa_input_before(cause, input, data) register_fsa_input_adv(cause, input, data, A_NOTHING, TRUE, __FUNCTION__) #define register_fsa_input_later(cause, input, data) register_fsa_input_adv(cause, input, data, A_NOTHING, FALSE, __FUNCTION__) void delete_fsa_input(fsa_data_t *fsa_data); GListPtr put_message(fsa_data_t *new_message); fsa_data_t *get_message(void); gboolean is_message(void); gboolean have_wait_message(void); extern gboolean relay_message( HA_Message *relay_message, gboolean originated_locally); extern void crmd_ha_msg_callback(const HA_Message * msg, void* private_data); extern gboolean crmd_ipc_msg_callback(IPC_Channel *client, gpointer user_data); extern void process_message( HA_Message *msg, gboolean originated_locally,const char *src_node_name); extern gboolean crm_dc_process_message(crm_data_t *whole_message, crm_data_t *action, const char *host_from, const char *sys_from, const char *sys_to, const char *op, gboolean dc_mode); extern gboolean send_msg_via_ha(ll_cluster_t *hb_fd, HA_Message *msg); extern gboolean send_msg_via_ipc(HA_Message *msg, const char *sys); extern gboolean add_pending_outgoing_reply(const char *originating_node_name, const char *crm_msg_reference, const char *sys_to, const char *sys_from); extern gboolean crmd_authorize_message( ha_msg_input_t *client_msg, crmd_client_t *curr_client); extern gboolean send_request(HA_Message *msg, char **msg_reference); extern enum crmd_fsa_input handle_message(ha_msg_input_t *stored_msg); extern gboolean send_ha_reply(ll_cluster_t *hb_cluster, crm_data_t *xml_request, crm_data_t *xml_response_data); extern void lrm_op_callback(lrm_op_t* op); extern void msg_queue_helper(void); #endif diff --git a/crm/crmd/election.c b/crm/crmd/election.c index 82885044ed..051e34ee3d 100644 --- a/crm/crmd/election.c +++ b/crm/crmd/election.c @@ -1,427 +1,425 @@ /* * Copyright (C) 2004 Andrew Beekhof * * 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.1 of the License, or (at your option) any later version. * * This software 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 library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include #include #include #include #include #include uint highest_born_on = -1; void ghash_count_vote(gpointer key, gpointer value, gpointer user_data); void revision_check_callback(const HA_Message *msg, int call_id, int rc, crm_data_t *output, void *user_data); /* A_ELECTION_VOTE */ enum crmd_fsa_input do_election_vote(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { gboolean not_voting = FALSE; HA_Message *vote = NULL; /* dont vote if we're in one of these states or wanting to shut down */ switch(cur_state) { case S_RECOVERY: case S_STOPPING: case S_TERMINATE: crm_warn("Not voting in election, we're in state %s", fsa_state2string(cur_state)); not_voting = TRUE; break; default: break; } if(not_voting == FALSE) { if(is_set(fsa_input_register, R_STARTING)) { not_voting = TRUE; } } if(not_voting) { fsa_cib_conn->cmds->set_slave(fsa_cib_conn, cib_scope_local); if(AM_I_DC) { register_fsa_input(C_FSA_INTERNAL, I_RELEASE_DC, NULL); } else { register_fsa_input(C_FSA_INTERNAL, I_PENDING, NULL); } return I_NULL; } vote = create_request( CRM_OP_VOTE, NULL, NULL, CRM_SYSTEM_CRMD, CRM_SYSTEM_CRMD, NULL); if(is_set(fsa_input_register, R_SHUTDOWN)) { crm_warn("Not voting in election, we're shutting down"); cl_msg_remove(vote, F_CRM_VERSION); } send_request(vote, NULL); crm_timer_start(election_timeout); return I_NULL; } char *dc_hb_msg = NULL; int beat_num = 0; gboolean do_dc_heartbeat(gpointer data) { #if 0 fsa_timer_t *timer = (fsa_timer_t *)data; crm_devel("Sending DC Heartbeat %d", beat_num); HA_Message *msg = ha_msg_new(5); ha_msg_add(msg, F_TYPE, T_CRM); ha_msg_add(msg, F_SUBTYPE, XML_ATTR_REQUEST); ha_msg_add(msg, F_CRM_SYS_TO, CRM_SYSTEM_CRMD); ha_msg_add(msg, F_CRM_SYS_FROM, CRM_SYSTEM_DC); ha_msg_add(msg, F_CRM_TASK, CRM_OP_HBEAT); ha_msg_add_int(msg, "dc_beat_seq", beat_num); beat_num++; if(send_msg_via_ha(fsa_cluster_conn, msg) == FALSE) { /* this is bad */ crm_timer_stop(timer); /* make it not go off again */ register_fsa_input(C_HEARTBEAT_FAILED, I_SHUTDOWN, NULL); - G_main_set_trigger(fsa_source); - return FALSE; } #endif return TRUE; } struct election_data_s { const char *winning_uname; unsigned int winning_bornon; }; /* A_ELECTION_COUNT */ enum crmd_fsa_input do_election_count_vote(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { gboolean we_loose = FALSE; ha_msg_input_t *vote = fsa_typed_data(fsa_dt_ha_msg); enum crmd_fsa_input election_result = I_NULL; const char *vote_from = cl_get_string(vote->msg, F_CRM_HOST_FROM); const char *your_version = cl_get_string(vote->msg, F_CRM_VERSION); oc_node_t *our_node = NULL, * your_node = NULL; if(vote_from == NULL || strcmp(vote_from, fsa_our_uname) == 0) { /* dont count our own vote */ return election_result; } if(fsa_membership_copy == NULL) { /* if the membership copy is NULL we REALLY shouldnt be voting * the question is how we managed to get here. */ crm_err("Membership copy was NULL"); return I_NULL; } else if(fsa_membership_copy->members != NULL) { our_node = (oc_node_t*) g_hash_table_lookup(fsa_membership_copy->members,fsa_our_uname); your_node = (oc_node_t*) g_hash_table_lookup(fsa_membership_copy->members,vote_from); } if(your_node == NULL) { crm_debug("Election ignore: The other side doesnt exist in CCM."); return I_NULL; /* if your_version == 0, then they're shutting down too */ } else if(is_set(fsa_input_register, R_SHUTDOWN)) { if(your_version != NULL) { crm_info("Election fail: we are shutting down"); we_loose = TRUE; } else { /* pretend nothing happened, they want to shutdown too*/ crm_info("Election ignore: they are shutting down too"); return I_NULL; } } else if(our_node == NULL || fsa_membership_copy->last_event == OC_EV_MS_EVICTED) { crm_info("Election fail: we dont exist in CCM"); we_loose = TRUE; } else if(your_version == NULL) { crm_info("Election pass: they are shutting down"); } else if(compare_version(your_version, CRM_VERSION) > 0) { crm_debug("Election fail: version"); we_loose = TRUE; } else if(compare_version(your_version, CRM_VERSION) < 0) { crm_debug("Election pass: version"); } else if(your_node->node_born_on < our_node->node_born_on) { crm_debug("Election fail: born_on"); we_loose = TRUE; } else if(your_node->node_born_on > our_node->node_born_on) { crm_debug("Election pass: born_on"); } else if(strcmp(fsa_our_uname, vote_from) > 0) { crm_debug("Election fail: uname"); we_loose = TRUE; /* cant happen... * } else if(strcmp(fsa_our_uname, vote_from) == 0) { * * default... * } else { // strcmp(fsa_our_uname, vote_from) < 0 * we win */ } if(we_loose) { crm_timer_stop(election_timeout); fsa_cib_conn->cmds->set_slave(fsa_cib_conn, cib_scope_local); crm_debug("Election lost to %s", vote_from); if(fsa_input_register & R_THE_DC) { crm_devel("Give up the DC to %s", vote_from); election_result = I_RELEASE_DC; } else { crm_devel("We werent the DC anyway"); election_result = I_PENDING; } } else { if(cur_state == S_PENDING) { crm_debug("Election ignore: We already lost the election"); return I_NULL; } else { crm_info("Election won over %s", vote_from); election_result = I_ELECTION; } } register_fsa_input(C_FSA_INTERNAL, election_result, NULL); return I_NULL; } /* A_ELECT_TIMER_START, A_ELECTION_TIMEOUT */ /* we won */ enum crmd_fsa_input do_election_timer_ctrl(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { return I_NULL; } /* A_DC_TAKEOVER */ enum crmd_fsa_input do_dc_takeover(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { crm_data_t *cib = createEmptyCib(); crm_data_t *update = NULL; crm_data_t *output = NULL; int rc = cib_ok; crm_info("Taking over DC status for this partition"); set_bit_inplace(fsa_input_register, R_THE_DC); crm_free(fsa_our_dc); fsa_our_dc = crm_strdup(fsa_our_uname); set_bit_inplace(fsa_input_register, R_JOIN_OK); set_bit_inplace(fsa_input_register, R_INVOKE_PE); if(dc_heartbeat->source_id == (guint)-1 || dc_heartbeat->source_id == (guint)-2) { crm_devel("Starting DC Heartbeat timer"); dc_heartbeat->source_id = Gmain_timeout_add_full( G_PRIORITY_HIGH, dc_heartbeat->period_ms, dc_heartbeat->callback, dc_heartbeat, NULL); } else { crm_devel("DC Heartbeat timer already active"); } /* fsa_cib_conn->cmds->set_slave_all(fsa_cib_conn, cib_none); */ fsa_cib_conn->cmds->set_master(fsa_cib_conn, cib_none); set_uuid(fsa_cluster_conn, cib, XML_ATTR_DC_UUID, fsa_our_uname); crm_devel("Update %s in the CIB to our uuid: %s", XML_ATTR_DC_UUID, crm_element_value(cib, XML_ATTR_DC_UUID)); update = create_cib_fragment(cib, NULL); free_xml(cib); rc = fsa_cib_conn->cmds->modify( fsa_cib_conn, NULL, update, &output, cib_quorum_override); if(rc == cib_revision_unsupported) { crm_err("Feature revision not permitted"); /* go into a stall state */ register_fsa_error(C_FSA_INTERNAL, I_HALT, NULL); return I_NULL; } else if(rc < cib_ok) { crm_err("DC UUID update failed: %s", cib_error2string(rc)); register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL); return I_NULL; } rc = fsa_cib_conn->cmds->query( fsa_cib_conn, NULL, NULL, cib_scope_local); if(FALSE == add_cib_op_callback( rc, TRUE, NULL, revision_check_callback)) { crm_err("Retrieval of generation failed"); register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL); return I_NULL; } free_xml(update); crm_devel("Requesting an initial dump of CRMD client_status"); fsa_cluster_conn->llc_ops->client_status( fsa_cluster_conn, NULL, CRM_SYSTEM_CRMD, -1); return I_NULL; } void revision_check_callback(const HA_Message *msg, int call_id, int rc, crm_data_t *output, void *user_data) { int revision_i = -1; const char *revision = NULL; crm_data_t *generation = find_xml_node(output, XML_TAG_CIB, TRUE); crm_devel("Checking our feature revision is allowed: %d", cib_feature_revision); revision = crm_element_value(generation, XML_ATTR_CIB_REVISION); revision_i = atoi(revision?revision:"0"); if(revision_i > cib_feature_revision) { crm_err("Feature revision not permitted"); /* go into a stall state */ register_fsa_error_adv( C_FSA_INTERNAL, I_HALT, NULL, NULL, __FUNCTION__); } } /* A_DC_RELEASE */ enum crmd_fsa_input do_dc_release(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { enum crmd_fsa_input result = I_NULL; crm_trace("################## Releasing the DC ##################"); crm_timer_stop(dc_heartbeat); if(action & A_DC_RELEASE) { clear_bit_inplace(fsa_input_register, R_THE_DC); } else if (action & A_DC_RELEASED) { fsa_cib_conn->cmds->set_slave(fsa_cib_conn, cib_scope_local); if(cur_state == S_STOPPING) { register_fsa_input(C_FSA_INTERNAL, I_RELEASE_SUCCESS, NULL); } #if 0 else if( are there errors ) { /* we cant stay up if not healthy */ /* or perhaps I_ERROR and go to S_RECOVER? */ result = I_SHUTDOWN; } #endif else { register_fsa_input(C_FSA_INTERNAL, I_RELEASE_SUCCESS, NULL); } } else { crm_err("Warning, do_dc_release invoked for action %s", fsa_action2string(action)); } crm_verbose("Am I still the DC? %s", AM_I_DC?XML_BOOLEAN_YES:XML_BOOLEAN_NO); return result; } void ghash_count_vote(gpointer key, gpointer value, gpointer user_data) { struct election_data_s *election_data = (struct election_data_s *)user_data; oc_node_t *cur_node = (oc_node_t*)value; const char *node_uname = (const char*)key; if(election_data->winning_bornon > cur_node->node_born_on) { election_data->winning_uname = node_uname; election_data->winning_bornon = cur_node->node_born_on; } else if(election_data->winning_bornon == cur_node->node_born_on && (election_data->winning_uname == NULL || strcmp(election_data->winning_uname, node_uname) > 0)) { election_data->winning_uname = node_uname; election_data->winning_bornon = cur_node->node_born_on; } } diff --git a/crm/crmd/messages.c b/crm/crmd/messages.c index 52be88b773..38e08de395 100644 --- a/crm/crmd/messages.c +++ b/crm/crmd/messages.c @@ -1,1206 +1,1210 @@ /* * Copyright (C) 2004 Andrew Beekhof * * 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.1 of the License, or (at your option) any later version. * * This software 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 library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include #include #include #include #include #include #include #include GListPtr fsa_message_queue = NULL; extern void crm_shutdown(int nsig); enum crmd_fsa_input handle_request(ha_msg_input_t *stored_msg); enum crmd_fsa_input handle_response(ha_msg_input_t *stored_msg); enum crmd_fsa_input handle_shutdown_request(HA_Message *stored_msg); ha_msg_input_t *copy_ha_msg_input(ha_msg_input_t *orig); gboolean ipc_queue_helper(gpointer key, gpointer value, gpointer user_data); #ifdef MSG_LOG # define ROUTER_RESULT(x) crm_devel("Router result: %s", x); \ crm_log_message_adv(LOG_MSG, "router.log", relay_message); #else # define ROUTER_RESULT(x) crm_devel("Router result: %s", x) #endif /* debug only, can wrap all it likes */ int last_data_id = 0; void register_fsa_error_adv( enum crmd_fsa_cause cause, enum crmd_fsa_input input, fsa_data_t *cur_data, void *new_data, const char *raised_from) { /* save the current actions */ register_fsa_input_adv(cur_data?cur_data->fsa_cause:C_FSA_INTERNAL, I_NULL, cur_data?cur_data->data:NULL, fsa_actions, TRUE, __FUNCTION__); /* reset the action list */ fsa_actions = A_NOTHING; /* register the error */ register_fsa_input_adv( cause, input, new_data, A_NOTHING, TRUE, raised_from); } static gboolean last_was_vote = FALSE; void register_fsa_input_adv( enum crmd_fsa_cause cause, enum crmd_fsa_input input, void *data, long long with_actions, gboolean prepend, const char *raised_from) { unsigned old_len = g_list_length(fsa_message_queue); fsa_data_t *fsa_data = NULL; crm_debug("%s raised FSA input %s (cause=%s) %s data", raised_from,fsa_input2string(input), fsa_cause2string(cause), data?"with":"without"); if(input == I_WAIT_FOR_EVENT) { do_fsa_stall = TRUE; set_bit_inplace(fsa_actions, with_actions); with_actions = A_NOTHING; crm_debug("Stalling the FSA pending further input"); if(old_len > 0) { crm_err("Stalling the FSA with pending inputs"); } fsa_dump_queue(LOG_DEBUG); return; } if(old_len == 0) { last_was_vote = FALSE; } if(input == I_NULL && with_actions == A_NOTHING /* && data == NULL */){ /* no point doing anything */ crm_err("Cannot add entry to queue: no input and no action"); return; } else if(data == NULL) { last_was_vote = FALSE; } else if(last_was_vote && cause == C_HA_MESSAGE && input == I_ROUTER) { const char *op = cl_get_string( ((ha_msg_input_t*)data)->msg, F_CRM_TASK); if(safe_str_eq(op, CRM_OP_VOTE)) { /* It is always safe to treat N successive votes as * a single one * * If all the discarded votes are more "loosing" than * the first then the result is accurate * (win or loose). * * If any of the discarded votes are less "loosing" * than the first then we will cast our vote and the * eventual winner will vote us down again (which * even in the case that N=2, is no worse than if we * had not disarded the vote). */ crm_verbose("Vote compression: %d", old_len); return; } } else if (cause == C_HA_MESSAGE && input == I_ROUTER) { const char *op = cl_get_string( ((ha_msg_input_t*)data)->msg, F_CRM_TASK); if(safe_str_eq(op, CRM_OP_VOTE)) { last_was_vote = TRUE; crm_devel("Added vote: %d", old_len); } } else { last_was_vote = FALSE; } crm_malloc0(fsa_data, sizeof(fsa_data_t)); fsa_data->id = ++last_data_id; fsa_data->fsa_input = input; fsa_data->fsa_cause = cause; fsa_data->origin = raised_from; fsa_data->data = NULL; fsa_data->data_type = fsa_dt_none; fsa_data->actions = with_actions; if(with_actions != A_NOTHING) { crm_devel("Adding actions %.16llx to input", with_actions); } if(data != NULL) { switch(cause) { case C_FSA_INTERNAL: case C_CRMD_STATUS_CALLBACK: case C_IPC_MESSAGE: case C_HA_MESSAGE: crm_devel("Copying %s data from %s as a HA msg", fsa_cause2string(cause), raised_from); fsa_data->data = copy_ha_msg_input(data); fsa_data->data_type = fsa_dt_ha_msg; break; case C_LRM_OP_CALLBACK: crm_devel("Copying %s data from %s as lrm_op_t", fsa_cause2string(cause), raised_from); fsa_data->data = copy_lrm_op((lrm_op_t*)data); fsa_data->data_type = fsa_dt_lrm; break; case C_CCM_CALLBACK: crm_devel("Copying %s data from %s as CCM data", fsa_cause2string(cause), raised_from); fsa_data->data = copy_ccm_data(data); fsa_data->data_type = fsa_dt_ccm; break; case C_SUBSYSTEM_CONNECT: case C_LRM_MONITOR_CALLBACK: case C_TIMER_POPPED: case C_SHUTDOWN: case C_HEARTBEAT_FAILED: case C_HA_DISCONNECT: case C_ILLEGAL: case C_UNKNOWN: case C_STARTUP: crm_err("Copying %s data (from %s)" " not yet implemented", fsa_cause2string(cause), raised_from); exit(1); break; } crm_trace("%s data copied", fsa_cause2string(fsa_data->fsa_cause)); } /* make sure to free it properly later */ if(prepend) { crm_trace("Prepending input"); fsa_message_queue = g_list_prepend(fsa_message_queue, fsa_data); } else { crm_trace("Appending input"); fsa_message_queue = g_list_append(fsa_message_queue, fsa_data); } crm_debug("Queue len: %d -> %d", old_len, g_list_length(fsa_message_queue)); fsa_dump_queue(LOG_DEBUG); if(old_len == g_list_length(fsa_message_queue)){ crm_err("Couldnt add message to the queue"); - } + } + if(fsa_source) { + G_main_set_trigger(fsa_source); + } } + void fsa_dump_queue(int log_level) { if(log_level < (int)crm_log_level) { return; } slist_iter( data, fsa_data_t, fsa_message_queue, lpc, do_crm_log(log_level, __FILE__, __FUNCTION__, "queue[%d(%d)]: input %s raised by %s()\t(cause=%s)", lpc, data->id, fsa_input2string(data->fsa_input), data->origin, fsa_cause2string(data->fsa_cause)); ); } ha_msg_input_t * copy_ha_msg_input(ha_msg_input_t *orig) { ha_msg_input_t *input_copy = NULL; crm_malloc0(input_copy, sizeof(ha_msg_input_t)); if(orig != NULL) { crm_trace("Copy msg"); input_copy->msg = ha_msg_copy(orig->msg); if(orig->xml != NULL) { crm_trace("Copy xml"); input_copy->xml = copy_xml_node_recursive(orig->xml); } } else { crm_devel("No message to copy"); } return input_copy; } void delete_fsa_input(fsa_data_t *fsa_data) { lrm_op_t *op = NULL; crm_data_t *foo = NULL; struct crmd_ccm_data_s *ccm_input = NULL; if(fsa_data == NULL) { return; } crm_trace("About to free %s data", fsa_cause2string(fsa_data->fsa_cause)); if(fsa_data->data != NULL) { switch(fsa_data->data_type) { case fsa_dt_ha_msg: delete_ha_msg_input(fsa_data->data); break; case fsa_dt_xml: foo = fsa_data->data; free_xml(foo); break; case fsa_dt_lrm: op = (lrm_op_t*)fsa_data->data; crm_free(op->user_data); crm_free(op->output); crm_free(op->rsc_id); crm_free(op->app_name); crm_free(op); break; case fsa_dt_ccm: ccm_input = (struct crmd_ccm_data_s *) fsa_data->data; crm_free(ccm_input->oc); crm_free(ccm_input); break; case fsa_dt_none: if(fsa_data->data != NULL) { crm_err("Dont know how to free %s data from %s", fsa_cause2string(fsa_data->fsa_cause), fsa_data->origin); exit(1); } break; } crm_trace("%s data freed", fsa_cause2string(fsa_data->fsa_cause)); } crm_free(fsa_data); } /* returns the next message */ fsa_data_t * get_message(void) { fsa_data_t* message = g_list_nth_data(fsa_message_queue, 0); fsa_message_queue = g_list_remove(fsa_message_queue, message); return message; } /* returns the current head of the FIFO queue */ gboolean is_message(void) { return (g_list_length(fsa_message_queue) > 0); } void * fsa_typed_data_adv( fsa_data_t *fsa_data, enum fsa_data_type a_type, const char *caller) { void *ret_val = NULL; if(fsa_data == NULL) { do_crm_log(LOG_ERR, caller, NULL, "No FSA data available"); } else if(fsa_data->data == NULL) { do_crm_log(LOG_ERR, caller, NULL, "No message data available"); } else if(fsa_data->data_type != a_type) { do_crm_log(LOG_CRIT, caller, NULL, "Message data was the wrong type! %d vs. requested=%d." " Origin: %s", fsa_data->data_type, a_type, fsa_data->origin); CRM_ASSERT(fsa_data->data_type == a_type); } else { ret_val = fsa_data->data; } return ret_val; } /* A_MSG_ROUTE */ enum crmd_fsa_input do_msg_route(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { enum crmd_fsa_input result = I_NULL; ha_msg_input_t *input = fsa_typed_data(fsa_dt_ha_msg); gboolean routed = FALSE; if(msg_data->fsa_cause != C_IPC_MESSAGE && msg_data->fsa_cause != C_HA_MESSAGE) { /* dont try and route these */ crm_warn("Can only process HA and IPC messages"); return I_NULL; } /* try passing the buck first */ crm_trace("Attempting to route message"); routed = relay_message(input->msg, cause==C_IPC_MESSAGE); if(routed == FALSE) { crm_trace("Message wasn't routed... try handling locally"); /* calculate defer */ result = handle_message(input); switch(result) { case I_NULL: break; case I_DC_HEARTBEAT: break; case I_CIB_OP: break; /* what else should go here? */ default: crm_trace("Defering local processing of message"); register_fsa_input_later( cause, result, msg_data->data); result = I_NULL; break; } if(result == I_NULL) { crm_trace("Message processed"); } else { register_fsa_input(cause, result, msg_data->data); } } else { crm_trace("Message routed..."); input->msg = NULL; } return I_NULL; } /* * This method frees msg */ gboolean send_request(HA_Message *msg, char **msg_reference) { gboolean was_sent = FALSE; /* crm_xml_devel(request, "Final request..."); */ if(msg_reference != NULL) { *msg_reference = crm_strdup( cl_get_string(msg, XML_ATTR_REFERENCE)); } was_sent = relay_message(msg, TRUE); if(was_sent == FALSE) { ha_msg_input_t *fsa_input = new_ha_msg_input(msg); register_fsa_input(C_IPC_MESSAGE, I_ROUTER, fsa_input); delete_ha_msg_input(fsa_input); crm_msg_del(msg); } return was_sent; } /* unless more processing is required, relay_message is freed */ gboolean relay_message(HA_Message *relay_message, gboolean originated_locally) { int is_for_dc = 0; int is_for_dcib = 0; int is_for_crm = 0; int is_for_cib = 0; int is_local = 0; gboolean processing_complete = FALSE; const char *host_to = cl_get_string(relay_message, F_CRM_HOST_TO); const char *sys_to = cl_get_string(relay_message, F_CRM_SYS_TO); const char *sys_from= cl_get_string(relay_message, F_CRM_SYS_FROM); const char *type = cl_get_string(relay_message, F_TYPE); const char *msg_error = NULL; crm_devel("Routing message %s", cl_get_string(relay_message, XML_ATTR_REFERENCE)); if(relay_message == NULL) { msg_error = "Cannot route empty message"; } else if(safe_str_eq(CRM_OP_HELLO, cl_get_string(relay_message, F_CRM_TASK))){ /* quietly ignore */ processing_complete = TRUE; } else if(safe_str_neq(type, T_CRM)) { msg_error = "Bad message type"; } else if(sys_to == NULL) { msg_error = "Bad message destination: no subsystem"; } if(msg_error != NULL) { processing_complete = TRUE; crm_err("%s", msg_error); crm_log_message(LOG_WARNING, relay_message); } if(processing_complete) { crm_msg_del(relay_message); return TRUE; } processing_complete = 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(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 && safe_str_neq(sys_from, CRM_SYSTEM_PENGINE) && safe_str_neq(sys_from, CRM_SYSTEM_TENGINE)) { /* Neither the TE or PE should be sending messages * to DC's on other nodes * * By definition, if we are no longer the DC, then * the PE or TE's data should be discarded */ ROUTER_RESULT("Message result: External relay to DC"); send_msg_via_ha(fsa_cluster_conn, relay_message); } else { /* discard */ ROUTER_RESULT("Message result: Discard, not DC"); crm_msg_del(relay_message); } } else if(is_local && (is_for_crm || is_for_cib)) { ROUTER_RESULT("Message result: CRMd process"); processing_complete = FALSE; /* more to be done by caller */ } else if(is_local) { ROUTER_RESULT("Message result: Local relay"); send_msg_via_ipc(relay_message, sys_to); } else { ROUTER_RESULT("Message result: External relay"); send_msg_via_ha(fsa_cluster_conn, relay_message); } return processing_complete; } gboolean crmd_authorize_message(ha_msg_input_t *client_msg, crmd_client_t *curr_client) { /* check the best case first */ const char *sys_from = cl_get_string(client_msg->msg, F_CRM_SYS_FROM); char *uuid = NULL; char *client_name = NULL; char *major_version = NULL; char *minor_version = NULL; const char *filtered_from; gpointer table_key = NULL; gboolean auth_result = FALSE; struct crm_subsystem_s *the_subsystem = NULL; gboolean can_reply = FALSE; /* no-one has registered with this id */ const char *op = cl_get_string(client_msg->msg, F_CRM_TASK); if (safe_str_neq(CRM_OP_HELLO, op)) { if(sys_from == NULL) { crm_warn("Message [%s] was had no value for %s... discarding", cl_get_string(client_msg->msg, XML_ATTR_REFERENCE), F_CRM_SYS_FROM); return FALSE; } 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_verbose("Message reply can%s be routed from %s.", can_reply?"":" not", sys_from); if(can_reply == FALSE) { crm_warn("Message [%s] not authorized", cl_get_string(client_msg->msg, XML_ATTR_REFERENCE)); } #if 0 if(ha_msg_value(msg, XML_ATTR_REFERENCE) == NULL) { ha_msg_add(new_input->msg, XML_ATTR_REFERENCE, seq); } #endif register_fsa_input(C_IPC_MESSAGE, I_ROUTER, client_msg); 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_warn("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_warn("Bit\t%.16llx not set in %.16llx", the_subsystem->flag_connected, fsa_input_register); crm_warn("Client %s joined but we dont need it", 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_verbose("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_devel("Updated client list with %s", crm_str(table_key)); if(the_subsystem != NULL) { set_bit_inplace( fsa_input_register, the_subsystem->flag_connected); } G_main_set_trigger(fsa_source); } else { crm_warn("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_NOOP) == 0) { crm_debug("no-op from %s", crm_str(host_from)); } else if(strcmp(op, CRM_OP_VOTE) == 0) { /* count the vote and decide what to do after that */ register_fsa_input_adv(C_HA_MESSAGE, I_NULL, stored_msg, A_ELECTION_COUNT, FALSE, __FUNCTION__); /* 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)); crm_info("Current state: %s", 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_JOIN_OFFER) == 0) { next_input = I_JOIN_OFFER; } else if(strcmp(op, CRM_OP_JOIN_ACKNAK) == 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_STOP; } 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(safe_str_eq(op, CRM_OP_TEABORT)) { if(fsa_state == S_POLICY_ENGINE || fsa_state == S_TRANSITION_ENGINE || fsa_state == S_IDLE) { next_input = I_PE_CALC; } else { crm_debug("Filtering %s op in state %s", op, fsa_state2string(fsa_state)); } } else if(safe_str_eq(op, CRM_OP_TETIMEOUT)) { if(fsa_state == S_TRANSITION_ENGINE || fsa_state == S_POLICY_ENGINE) { next_input = I_PE_CALC; } else if(fsa_state == S_IDLE) { crm_err("Transition timed out in S_IDLE"); next_input = I_PE_CALC; } else { crm_err("Filtering %s op in state %s", 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_debug("Filtering %s op in state %s", op, fsa_state2string(fsa_state)); } } else if(strcmp(op, CRM_OP_JOIN_ANNOUNCE) == 0) { next_input = I_NODE_JOIN; } else if(strcmp(op, CRM_OP_JOIN_REQUEST) == 0) { next_input = I_JOIN_REQUEST; } else if(strcmp(op, CRM_OP_JOIN_CONFIRM) == 0) { next_input = I_JOIN_RESULT; } 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" " (DC) is askin us to shutdown!"); next_input = I_STOP; } else if(fsa_state != S_STOPPING) { crm_err("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_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_SHUTDOWN_REQ) == 0 || strcmp(op, CRM_OP_SHUTDOWN) == 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 */ } 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_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_quorum_override); free_xml(frag); /* will be picked up by the TE as long as its running */ if(need_transition(fsa_state) && is_set(fsa_input_register, R_TE_CONNECTED) == FALSE) { register_fsa_action(A_TE_CANCEL); } 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; 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; all_is_good = send_ha_message(hb_fd, msg, NULL); } else { all_is_good = send_ha_message(hb_fd, msg, host_to); } } 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, __FILE__, __FUNCTION__, "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?"":host_to, all_is_good?"succeeded":"failed"); } 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_malloc0(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; } void msg_queue_helper(void) { IPC_Channel *ipc = NULL; if(fsa_cluster_conn != NULL) { ipc = fsa_cluster_conn->llc_ops->ipcchan( fsa_cluster_conn); } if(ipc != NULL) { ipc->ops->resume_io(ipc); } /* g_hash_table_foreach_remove(ipc_clients, ipc_queue_helper, NULL); */ } gboolean ipc_queue_helper(gpointer key, gpointer value, gpointer user_data) { crmd_client_t *ipc_client = value; if(ipc_client->client_channel != NULL) { ipc_client->client_channel->ops->is_message_pending(ipc_client->client_channel); } return FALSE; } diff --git a/crm/crmd/pengine.c b/crm/crmd/pengine.c index c2c363040f..6e1f43a875 100644 --- a/crm/crmd/pengine.c +++ b/crm/crmd/pengine.c @@ -1,215 +1,214 @@ /* * Copyright (C) 2004 Andrew Beekhof * * 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.1 of the License, or (at your option) any later version. * * This software 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 library; if not, write to the Free Software * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include #include #include #include #include #include /* for access */ #include #include #include /* for calls to open */ #include /* for calls to open */ #include /* for calls to open */ #include /* for getpwuid */ #include /* for initgroups */ #include /* for getrlimit */ #include /* for getrlimit */ #include #include #include #include #include #include #include #include #define CLIENT_EXIT_WAIT 30 struct crm_subsystem_s *pe_subsystem = NULL; void do_pe_invoke_callback(const HA_Message *msg, int call_id, int rc, crm_data_t *output, void *user_data); /* A_PE_START, A_PE_STOP, A_TE_RESTART */ enum crmd_fsa_input do_pe_control(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { enum crmd_fsa_input result = I_NULL; struct crm_subsystem_s *this_subsys = pe_subsystem; long long stop_actions = A_PE_STOP; long long start_actions = A_PE_START; if(action & stop_actions) { if(stop_subsystem(this_subsys) == FALSE) { register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL); } } if(action & start_actions) { if(cur_state != S_STOPPING) { if(start_subsystem(this_subsys) == FALSE) { register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL); cleanup_subsystem(this_subsys); } } else { crm_info("Ignoring request to start %s while shutting down", this_subsys->name); } } return result; } char *fsa_pe_ref = NULL; /* A_PE_INVOKE */ enum crmd_fsa_input do_pe_invoke(long long action, enum crmd_fsa_cause cause, enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t *msg_data) { int call_id = 0; /* * FIXME: The CIB might have a different version of membership than the CRM * We need to allow for that possibility. * We could set a flag saying we're waiting for the membership versions (and quorum!) * to synchronize before going on. I don't know if anything bad happens * if the CIB is ahead of us. But I know for sure that bad things * happen when the CIB is behind us (the CRM). * * This probably has effects beyond that of running things without quorum * or failing to run things when we have quorum. * * We might try and run things on nodes that aren't running, and we * might fail to schedule something on a node which is really available * for use. I'm pretty sure I've seen the latter occur * * A crude method would be to poll every 100ms and detect when the CRM * and CIB membership versions are the same. I suspect if I knew * the code better, there probably is a callback which occurs when the * CIB is updated which we could use to trigger the delayed PE invocation. * There _might_ also need to be a mechanism for cancelling this delayed * pengine invocation - depending on what else happens after we * get this far (this doesn't seem that likely) * --AlanR. * */ if(is_set(fsa_input_register, R_PE_CONNECTED) == FALSE){ if(pe_subsystem->pid > 0) { int pid_status = -1; int rc = waitpid( pe_subsystem->pid, &pid_status, WNOHANG); if(rc > 0 && WIFEXITED(pid_status)) { clear_bit_inplace(fsa_input_register, pe_subsystem->flag_connected); if(is_set(fsa_input_register, pe_subsystem->flag_required)) { /* this wasnt supposed to happen */ crm_err("%s[%d] terminated during start", pe_subsystem->name, pe_subsystem->pid); register_fsa_error( C_FSA_INTERNAL, I_ERROR, NULL); } pe_subsystem->pid = -1; return I_NULL; } } crm_info("Waiting for the PE to connect"); crmd_fsa_stall(); return I_NULL; } crm_debug("Requesting the current CIB"); call_id = fsa_cib_conn->cmds->query( fsa_cib_conn, NULL, NULL, cib_scope_local); if(FALSE == add_cib_op_callback( call_id, TRUE, NULL, do_pe_invoke_callback)) { crm_err("Cant retrieve the CIB to invoke the %s subsystem with", pe_subsystem->name); register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL); } return I_NULL; } void do_pe_invoke_callback(const HA_Message *msg, int call_id, int rc, crm_data_t *output, void *user_data) { HA_Message *cmd = NULL; int ccm_transition_id = -1; gboolean cib_has_quorum = FALSE; crm_data_t *local_cib = find_xml_node(output, XML_TAG_CIB, TRUE); if(AM_I_DC == FALSE || is_set(fsa_input_register, R_PE_CONNECTED) == FALSE || fsa_state != S_POLICY_ENGINE) { crm_debug("No need to invoke the PE anymore"); return; } crm_verbose("Invoking %s with %p", CRM_SYSTEM_PENGINE, local_cib); CRM_DEV_ASSERT(local_cib != NULL); CRM_DEV_ASSERT(crm_element_value(local_cib, XML_ATTR_DC_UUID) != NULL); cib_has_quorum = crm_is_true( crm_element_value(local_cib, XML_ATTR_HAVE_QUORUM)); ccm_transition_id = crm_atoi( crm_element_value(local_cib, XML_ATTR_CCM_TRANSITION), "-1"); if(ccm_transition_id != fsa_membership_copy->id) { - crm_err("Re-asking for the CIB until membership/quorum" - " matches: CIB=%d, CRM=%d", - ccm_transition_id, fsa_membership_copy->id); + crm_debug("Re-asking for the CIB until membership/quorum" + " matches: CIB=%d, CRM=%d", + ccm_transition_id, fsa_membership_copy->id); register_fsa_action(A_PE_INVOKE); - G_main_set_trigger(fsa_source); return; } if(fsa_pe_ref) { crm_free(fsa_pe_ref); fsa_pe_ref = NULL; } cmd = create_request( CRM_OP_PECALC, local_cib, NULL, CRM_SYSTEM_PENGINE, CRM_SYSTEM_DC, NULL); send_request(cmd, &fsa_pe_ref); }