diff --git a/crm/ais/plugin.c b/crm/ais/plugin.c index d48d217d0a..a845af05cf 100644 --- a/crm/ais/plugin.c +++ b/crm/ais/plugin.c @@ -1,1170 +1,1227 @@ /* * 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 #include #include #include "plugin.h" #include "utils.h" #include #define OPENAIS_EXTERNAL_SERVICE insane_ais_header_hack_in__totem_h #include #include #include #include -#include - +#ifndef AIS_WHITETANK +# include +#endif #include #include #include #include #include #include #include int plugin_log_level = LOG_DEBUG; char *local_uname = NULL; int local_uname_len = 0; unsigned int local_nodeid = 0; char *ipc_channel_name = NULL; unsigned long long membership_seq = 0; pthread_t crm_wait_thread; gboolean wait_active = TRUE; GHashTable *membership_list = NULL; #define MAX_RESPAWN 100 #define crm_flag_none 0x00000000 #define crm_flag_members 0x00000001 struct crm_identify_msg_s { mar_req_header_t header __attribute__((aligned(8))); uint32_t id; uint32_t pid; int32_t votes; uint32_t processes; char uname[256]; char version[256]; } __attribute__((packed)); static crm_child_t crm_children[] = { - { 0, crm_proc_none, crm_flag_none, 0, FALSE, "none", 0, NULL, NULL }, - { 0, crm_proc_ais, crm_flag_none, 0, FALSE, "ais", 0, NULL, NULL }, - { 0, crm_proc_lrmd, crm_flag_none, 0, TRUE, "lrmd", 0, HA_LIBHBDIR"/lrmd", NULL }, - { 0, crm_proc_cib, crm_flag_members, 0, TRUE, "cib", HA_CCMUID, HA_LIBHBDIR"/cib", NULL }, - { 0, crm_proc_crmd, crm_flag_members, 0, TRUE, "crmd", HA_CCMUID, HA_LIBHBDIR"/crmd", NULL }, - { 0, crm_proc_attrd,crm_flag_none, 0, TRUE, "attrd", HA_CCMUID, HA_LIBHBDIR"/attrd", NULL }, - { 0, crm_proc_stonithd,crm_flag_none, 0, TRUE, "stonithd", 0, HA_LIBHBDIR"/stonithd", NULL }, + { 0, crm_proc_none, crm_flag_none, 0, 0, FALSE, "none", 0, NULL, NULL }, + { 0, crm_proc_ais, crm_flag_none, 0, 0, FALSE, "ais", 0, NULL, NULL }, + { 0, crm_proc_lrmd, crm_flag_none, 2, 0, TRUE, "lrmd", 0, HA_LIBHBDIR"/lrmd", NULL }, + { 0, crm_proc_cib, crm_flag_members, 1, 0, TRUE, "cib", HA_CCMUID, HA_LIBHBDIR"/cib", NULL }, + { 0, crm_proc_crmd, crm_flag_members, 5, 0, TRUE, "crmd", HA_CCMUID, HA_LIBHBDIR"/crmd", NULL }, + { 0, crm_proc_attrd, crm_flag_none, 4, 0, TRUE, "attrd", HA_CCMUID, HA_LIBHBDIR"/attrd", NULL }, + { 0, crm_proc_stonithd, crm_flag_none, 3, 0, TRUE, "stonithd", 0, HA_LIBHBDIR"/stonithd", NULL }, }; void send_cluster_id(void); int send_cluster_msg_raw(AIS_Message *ais_msg); char *ais_generate_membership_data(void); extern totempg_groups_handle openais_group_handle; void global_confchg_fn ( enum totem_configuration_type configuration_type, unsigned int *member_list, int member_list_entries, unsigned int *left_list, int left_list_entries, unsigned int *joined_list, int joined_list_entries, struct memb_ring_id *ring_id); int crm_exec_exit_fn (struct objdb_iface_ver0 *objdb); int crm_exec_init_fn (struct objdb_iface_ver0 *objdb); int crm_config_init_fn(struct objdb_iface_ver0 *objdb); int ais_ipc_client_connect_callback (void *conn); int ais_ipc_client_exit_callback (void *conn); void ais_cluster_message_swab(void *msg); void ais_cluster_message_callback(void *message, unsigned int nodeid); void ais_ipc_message_callback(void *conn, void *msg); void ais_quorum_query(void *conn, void *msg); void ais_node_list_query(void *conn, void *msg); void ais_manage_notification(void *conn, void *msg); void ais_cluster_id_swab(void *msg); void ais_cluster_id_callback(void *message, unsigned int nodeid); static struct openais_lib_handler crm_lib_service[] = { { /* 0 */ .lib_handler_fn = ais_ipc_message_callback, .response_size = sizeof (mar_res_header_t), .response_id = CRM_MESSAGE_IPC_ACK, .flow_control = OPENAIS_FLOW_CONTROL_NOT_REQUIRED }, { /* 1 */ .lib_handler_fn = ais_node_list_query, .response_size = sizeof (mar_res_header_t), .response_id = CRM_MESSAGE_IPC_ACK, .flow_control = OPENAIS_FLOW_CONTROL_NOT_REQUIRED }, { /* 2 */ .lib_handler_fn = ais_manage_notification, .response_size = sizeof (mar_res_header_t), .response_id = CRM_MESSAGE_IPC_ACK, .flow_control = OPENAIS_FLOW_CONTROL_NOT_REQUIRED }, }; static struct openais_exec_handler crm_exec_service[] = { { /* 0 */ .exec_handler_fn = ais_cluster_message_callback, .exec_endian_convert_fn = ais_cluster_message_swab }, { /* 1 */ .exec_handler_fn = ais_cluster_id_callback, .exec_endian_convert_fn = ais_cluster_id_swab } }; static void crm_exec_dump_fn(void) { ENTER(""); ais_err("Called after SIG_USR2"); LEAVE(""); } /* * Exports the interface for the service */ struct openais_service_handler crm_service_handler = { .name = "LHA Cluster Manager", .id = CRM_SERVICE, .private_data_size = 0, .flow_control = OPENAIS_FLOW_CONTROL_NOT_REQUIRED, .lib_init_fn = ais_ipc_client_connect_callback, .lib_exit_fn = ais_ipc_client_exit_callback, .lib_service = crm_lib_service, .lib_service_count = sizeof (crm_lib_service) / sizeof (struct openais_lib_handler), .exec_init_fn = crm_exec_init_fn, .exec_exit_fn = crm_exec_exit_fn, .exec_service = crm_exec_service, .exec_service_count = sizeof (crm_exec_service) / sizeof (struct openais_exec_handler), .config_init_fn = crm_config_init_fn, .confchg_fn = global_confchg_fn, .exec_dump_fn = crm_exec_dump_fn, /* void (*sync_init) (void); */ /* int (*sync_process) (void); */ /* void (*sync_activate) (void); */ /* void (*sync_abort) (void); */ }; /* * Dynamic Loader definition */ struct openais_service_handler *crm_get_handler_ver0 (void); static struct openais_service_handler_iface_ver0 crm_service_handler_iface = { .openais_get_service_handler_ver0 = crm_get_handler_ver0 }; static struct lcr_iface openais_crm_ver0[1] = { { .name = "lha_crm", .version = 0, .versions_replace = 0, .versions_replace_count = 0, .dependencies = 0, .dependency_count = 0, .constructor = NULL, .destructor = NULL, .interfaces = NULL } }; static struct lcr_comp crm_comp_ver0 = { .iface_count = 1, .ifaces = openais_crm_ver0 }; struct openais_service_handler *crm_get_handler_ver0 (void) { return (&crm_service_handler); } __attribute__ ((constructor)) static void register_this_component (void) { lcr_interfaces_set (&openais_crm_ver0[0], &crm_service_handler_iface); lcr_component_register (&crm_comp_ver0); } /* IMPL */ int crm_config_init_fn(struct objdb_iface_ver0 *objdb) { int rc = 0; struct utsname us; char *value = NULL; unsigned int object_service_handle; +#ifdef AIS_WHITETANK + log_init ("crm"); +#endif + membership_list = g_hash_table_new_full( g_direct_hash, g_direct_equal, NULL, destroy_ais_node); setenv("HA_COMPRESSION", "bz2", 1); setenv("HA_cluster_type", "openais", 1); objdb->object_find_reset (OBJECT_PARENT_HANDLE); if (objdb->object_find ( OBJECT_PARENT_HANDLE, "pacemaker", strlen ("pacemaker"), &object_service_handle) != 0) { object_service_handle = 0; ais_info("No configuration suplpied for pacemaker"); } objdb_get_string( objdb, object_service_handle, "logfacility", &value, "daemon"); setenv("HA_logfacility", value, 1); objdb_get_string(objdb, object_service_handle, "initdead", &value, "20"); setenv("HA_initdead", value, 1); objdb_get_string(objdb, object_service_handle, "debug", &value, "1"); setenv("HA_debug", value, 1); rc = atoi(value); plugin_log_level = LOG_INFO+rc; if(system("echo 1 > /proc/sys/kernel/core_uses_pid") != 0) { ais_perror("Could not enable /proc/sys/kernel/core_uses_pid"); } ais_info("CRM: Initialized"); log_printf(LOG_INFO, "Logging: Initialized %s\n", __PRETTY_FUNCTION__); rc = uname(&us); AIS_ASSERT(rc == 0); local_uname = ais_strdup(us.nodename); local_uname_len = strlen(local_uname); ais_info("Local hostname: %s", local_uname); local_nodeid = totempg_my_nodeid_get(); update_member(local_nodeid, 0, 1, 0, local_uname, CRM_NODE_LOST); LEAVE(""); return 0; } static void *crm_wait_dispatch (void *arg) { struct timespec waitsleep = { .tv_sec = 0, .tv_nsec = 100000 /* 100 msec */ }; while(wait_active) { int lpc = 0; for (; lpc < SIZEOF(crm_children); lpc++) { if(crm_children[lpc].pid > 0) { int status; pid_t pid = wait4( crm_children[lpc].pid, &status, WNOHANG, NULL); if(pid == 0) { continue; } else if(pid < 0) { ais_perror("crm_wait_dispatch: Call to wait4(%s) failed", crm_children[lpc].name); continue; } /* cleanup */ crm_children[lpc].pid = 0; crm_children[lpc].conn = NULL; crm_children[lpc].async_conn = NULL; if(WIFSIGNALED(status)) { int sig = WTERMSIG(status); ais_warn("Child process %s terminated with signal %d" " (pid=%d, core=%s)", crm_children[lpc].name, sig, pid, WCOREDUMP(status)?"true":"false"); } else if (WIFEXITED(status)) { int rc = WEXITSTATUS(status); ais_notice("Child process %s exited (pid=%d, rc=%d)", crm_children[lpc].name, pid, rc); if(rc == 100) { ais_notice("Child process %s no longer wishes" " to be respawned", crm_children[lpc].name); crm_children[lpc].respawn = FALSE; } } crm_children[lpc].respawn_count += 1; if(crm_children[lpc].respawn_count > MAX_RESPAWN) { ais_notice("Child respawn count exceeded by %s", crm_children[lpc].name); crm_children[lpc].respawn = FALSE; } if(crm_children[lpc].respawn) { ais_info("Respawning failed child process: %s", crm_children[lpc].name); spawn_child(&(crm_children[lpc])); } else { send_cluster_id(); } } } sched_yield (); nanosleep (&waitsleep, 0); } return 0; } #include int crm_exec_init_fn (struct objdb_iface_ver0 *objdb) { int lpc = 0; + int start_seq = 1; + static int max = SIZEOF(crm_children); ENTER(""); mkdir(HA_VARRUNDIR, 750); mkdir(HA_VARRUNDIR"/crm", 750); chown(HA_VARRUNDIR"/crm", HA_CCMUID, HA_APIGID); chown(HA_VARRUNDIR, HA_CCMUID, HA_APIGID); pthread_create (&crm_wait_thread, NULL, crm_wait_dispatch, NULL); - for (; lpc < SIZEOF(crm_children); lpc++) { - spawn_child(&(crm_children[lpc])); + for (start_seq = 1; start_seq < max; start_seq++) { + /* dont start anything with start_seq < 1 */ + for (lpc = 0; lpc < max; lpc++) { + if(start_seq == crm_children[lpc].start_seq) { + spawn_child(&(crm_children[lpc])); + } + } } ais_info("CRM: Initialized"); LEAVE(""); return 0; } /* static void ais_print_node(const char *prefix, struct totem_ip_address *host) { int len = 0; char *buffer = NULL; ais_malloc0(buffer, INET6_ADDRSTRLEN+1); inet_ntop(host->family, host->addr, buffer, INET6_ADDRSTRLEN); len = strlen(buffer); ais_info("%s: %.*s", prefix, len, buffer); ais_free(buffer); } */ #if 0 /* copied here for reference from exec/totempg.c */ char *totempg_ifaces_print (unsigned int nodeid) { static char iface_string[256 * INTERFACE_MAX]; char one_iface[64]; struct totem_ip_address interfaces[INTERFACE_MAX]; char **status; unsigned int iface_count; unsigned int i; int res; iface_string[0] = '\0'; res = totempg_ifaces_get (nodeid, interfaces, &status, &iface_count); if (res == -1) { return ("no interface found for nodeid"); } for (i = 0; i < iface_count; i++) { sprintf (one_iface, "r(%d) ip(%s) ", i, totemip_print (&interfaces[i])); strcat (iface_string, one_iface); } return (iface_string); } #endif static void ais_mark_unseen_peer_dead( gpointer key, gpointer value, gpointer user_data) { int *changed = user_data; crm_node_t *node = value; if(node->last_seen != membership_seq && ais_str_eq(CRM_NODE_LOST, node->state) == FALSE) { ais_info("Node %s was not seen in the previous transition", node->uname); *changed += update_member(node->id, membership_seq, node->votes, node->processes, node->uname, CRM_NODE_LOST); ais_info("Node %s marked dead", node->uname); } } void global_confchg_fn ( enum totem_configuration_type configuration_type, unsigned int *member_list, int member_list_entries, unsigned int *left_list, int left_list_entries, unsigned int *joined_list, int joined_list_entries, struct memb_ring_id *ring_id) { int lpc = 0; int changed = 0; int do_update = 0; ENTER(""); AIS_ASSERT(ring_id != NULL); switch(configuration_type) { case TOTEM_CONFIGURATION_REGULAR: do_update = 1; break; case TOTEM_CONFIGURATION_TRANSITIONAL: break; } membership_seq = ring_id->seq; ais_notice("%s membership event on ring %lld: memb=%d, new=%d, lost=%d", do_update?"Stable":"Transitional", ring_id->seq, member_list_entries, joined_list_entries, left_list_entries); if(do_update == 0) { for(lpc = 0; lpc < joined_list_entries; lpc++) { const char *prefix = "new: "; uint32_t nodeid = joined_list[lpc]; ais_info("%s %s %u", prefix, member_uname(nodeid), nodeid); } for(lpc = 0; lpc < member_list_entries; lpc++) { const char *prefix = "memb:"; uint32_t nodeid = member_list[lpc]; ais_info("%s %s %u", prefix, member_uname(nodeid), nodeid); } for(lpc = 0; lpc < left_list_entries; lpc++) { const char *prefix = "lost:"; uint32_t nodeid = left_list[lpc]; ais_info("%s %s %u", prefix, member_uname(nodeid), nodeid); } return; } for(lpc = 0; lpc < joined_list_entries; lpc++) { const char *prefix = "NEW: "; uint32_t nodeid = joined_list[lpc]; crm_node_t *node = NULL; changed += update_member( nodeid, membership_seq, -1, 0, NULL, CRM_NODE_MEMBER); ais_info("%s %s %u", prefix, member_uname(nodeid), nodeid); node = g_hash_table_lookup(membership_list, GUINT_TO_POINTER(nodeid)); if(node->addr == NULL) { const char *addr = totempg_ifaces_print(nodeid); node->addr = ais_strdup(addr); ais_debug("Node %u has address %s", nodeid, node->addr); } } for(lpc = 0; lpc < member_list_entries; lpc++) { const char *prefix = "MEMB:"; uint32_t nodeid = member_list[lpc]; changed += update_member( nodeid, membership_seq, -1, 0, NULL, CRM_NODE_MEMBER); ais_info("%s %s %u", prefix, member_uname(nodeid), nodeid); } for(lpc = 0; lpc < left_list_entries; lpc++) { const char *prefix = "LOST:"; uint32_t nodeid = left_list[lpc]; changed += update_member( nodeid, membership_seq, -1, 0, NULL, CRM_NODE_LOST); ais_info("%s %s %u", prefix, member_uname(nodeid), nodeid); } if(do_update) { ais_debug_2("Reaping unseen nodes..."); g_hash_table_foreach( membership_list, ais_mark_unseen_peer_dead, &changed); } if(changed) { ais_debug("%d nodes changed", changed); send_member_notification(); } send_cluster_id(); LEAVE(""); } int ais_ipc_client_exit_callback (void *conn) { int lpc = 0; const char *client = NULL; ENTER("Client=%p", conn); for (; lpc < SIZEOF(crm_children); lpc++) { if(crm_children[lpc].conn == conn) { crm_children[lpc].conn = NULL; crm_children[lpc].async_conn = NULL; client = crm_children[lpc].name; break; } } ais_info("Client %p/%s left", conn, client?client:"unknown-transient"); LEAVE(""); return (0); } int ais_ipc_client_connect_callback (void *conn) { void *async_conn = openais_conn_partner_get(conn); ENTER("Client=%p", conn); ais_debug("Client %p/%p joined", conn, async_conn); +#ifndef AIS_WHITETANK + /* Whitetank hasn't finished setting up the connection at this point + * Sending messages now messes up the protocol! + */ if(async_conn) { send_client_msg(async_conn, crm_class_cluster, crm_msg_none, "identify"); } else { ais_err("No async connection"); } +#endif LEAVE(""); return (0); } /* * Executive message handlers */ void ais_cluster_message_swab(void *msg) { AIS_Message *ais_msg = msg; ENTER(""); ais_debug_3("Performing endian conversion..."); ais_msg->id = swab32 (ais_msg->id); ais_msg->is_compressed = swab32 (ais_msg->is_compressed); ais_msg->compressed_size = swab32 (ais_msg->compressed_size); ais_msg->size = swab32 (ais_msg->size); ais_msg->host.id = swab32 (ais_msg->host.id); ais_msg->host.pid = swab32 (ais_msg->host.pid); ais_msg->host.type = swab32 (ais_msg->host.type); ais_msg->host.size = swab32 (ais_msg->host.size); ais_msg->host.local = swab32 (ais_msg->host.local); ais_msg->sender.id = swab32 (ais_msg->sender.id); ais_msg->sender.pid = swab32 (ais_msg->sender.pid); ais_msg->sender.type = swab32 (ais_msg->sender.type); ais_msg->sender.size = swab32 (ais_msg->sender.size); ais_msg->sender.local = swab32 (ais_msg->sender.local); LEAVE(""); } void ais_cluster_message_callback ( void *message, unsigned int nodeid) { AIS_Message *ais_msg = message; ENTER("Node=%u (%s)", nodeid, nodeid==local_nodeid?"local":"remote"); ais_debug_2("Message from node %u (%s)", nodeid, nodeid==local_nodeid?"local":"remote"); /* Shouldn't be required... update_member( ais_msg->sender.id, membership_seq, -1, 0, ais_msg->sender.uname, NULL); */ if(ais_msg->host.size == 0 || ais_str_eq(ais_msg->host.uname, local_uname)) { route_ais_message(ais_msg, FALSE); } else { ais_debug_3("Discarding Msg[%d] (dest=%s:%s, from=%s:%s)", ais_msg->id, ais_dest(&(ais_msg->host)), msg_type2text(ais_msg->host.type), ais_dest(&(ais_msg->sender)), msg_type2text(ais_msg->sender.type)); } LEAVE(""); } void ais_cluster_id_swab(void *msg) { struct crm_identify_msg_s *ais_msg = msg; ENTER(""); ais_debug_3("Performing endian conversion..."); ais_msg->id = swab32 (ais_msg->id); ais_msg->pid = swab32 (ais_msg->pid); ais_msg->votes = swab32 (ais_msg->votes); ais_msg->processes = swab32 (ais_msg->processes); LEAVE(""); } void ais_cluster_id_callback (void *message, unsigned int nodeid) { int changed = 0; struct crm_identify_msg_s *msg = message; if(nodeid != msg->id) { ais_err("Invalid message: Node %u claimed to be node %d", nodeid, msg->id); return; } ais_debug("Node update: %s (%s)", msg->uname, msg->version); changed = update_member( nodeid, membership_seq, msg->votes, msg->processes, msg->uname, NULL); if(changed) { send_member_notification(); } } struct res_overlay { mar_res_header_t header __attribute((aligned(8))); char buf[4096]; }; struct res_overlay *res_overlay = NULL; static void send_ipc_ack(void *conn, int class) { if(res_overlay == NULL) { ais_malloc0(res_overlay, sizeof(struct res_overlay)); } res_overlay->header.size = crm_lib_service[class].response_size; res_overlay->header.id = crm_lib_service[class].response_id; - res_overlay->header.error = 0; + res_overlay->header.error = SA_AIS_OK; +#ifdef AIS_WHITETANK + openais_response_send (conn, res_overlay, res_overlay->header.size); +#else openais_conn_send_response (conn, res_overlay, res_overlay->header.size); +#endif } /* local callbacks */ void ais_ipc_message_callback(void *conn, void *msg) { + gboolean transient = TRUE; AIS_Message *ais_msg = msg; int type = ais_msg->sender.type; void *async_conn = openais_conn_partner_get(conn); ENTER("Client=%p", conn); ais_debug_2("Message from client %p", conn); + send_ipc_ack(conn, 0); - if(type > 0 + ais_debug_3("type: %d local: %d conn: %p host type: %d ais: %d sender pid: %d child pid: %d size: %d", + type, ais_msg->host.local, crm_children[type].conn, ais_msg->host.type, crm_msg_ais, + ais_msg->sender.pid, crm_children[type].pid, ((int)SIZEOF(crm_children))); + + if(type > crm_msg_none && type < SIZEOF(crm_children)) { + /* known child process */ + transient = FALSE; + } + + /* If this check fails, the order of crm_children probably + * doesn't match that of the crm_ais_msg_types enum + */ + AIS_CHECK(transient || ais_msg->sender.pid == crm_children[type].pid, + ais_err("Sender: %d, child[%d]: %d", ais_msg->sender.pid, type, crm_children[type].pid); + return); + + if(transient == FALSE + && type > crm_msg_none && ais_msg->host.local && crm_children[type].conn == NULL - && ais_msg->host.type == crm_msg_ais - && ais_msg->sender.pid == crm_children[type].pid - && type < SIZEOF(crm_children)) { + && ais_msg->host.type == crm_msg_ais) { + ais_info("Recorded connection %p for %s/%d", conn, crm_children[type].name, crm_children[type].pid); crm_children[type].conn = conn; crm_children[type].async_conn = async_conn; /* Make sure they have the latest membership */ if(crm_children[type].flags & crm_flag_members) { char *update = ais_generate_membership_data(); ais_info("Sending membership update %llu to %s", membership_seq, crm_children[type].name); send_client_msg(async_conn, crm_class_members, crm_msg_none,update); } } ais_msg->sender.id = local_nodeid; ais_msg->sender.size = local_uname_len; memset(ais_msg->sender.uname, 0, MAX_NAME); memcpy(ais_msg->sender.uname, local_uname, ais_msg->sender.size); route_ais_message(msg, TRUE); - send_ipc_ack(conn, 0); LEAVE(""); } int crm_exec_exit_fn (struct objdb_iface_ver0 *objdb) { int lpc = 0; + int start_seq = 1; + static int max = SIZEOF(crm_children); + struct timespec waitsleep = { .tv_sec = 1, .tv_nsec = 0 }; ENTER(""); ais_notice("Begining shutdown"); in_shutdown = TRUE; wait_active = FALSE; /* stop the wait loop */ - - for (lpc = SIZEOF(crm_children) - 1; lpc > 0; lpc--) { - crm_children[lpc].respawn = FALSE; - stop_child(&(crm_children[lpc]), SIGTERM); - while(crm_children[lpc].command && crm_children[lpc].pid) { - int status; - pid_t pid = 0; - - pid = wait4( - crm_children[lpc].pid, &status, WNOHANG, NULL); - - if(pid == 0) { - sched_yield (); - nanosleep (&waitsleep, 0); + + for (start_seq = max; start_seq > 0; start_seq--) { + /* dont stop anything with start_seq < 1 */ + + for (lpc = max - 1; lpc >= 0; lpc--) { + if(start_seq != crm_children[lpc].start_seq) { continue; + } + + crm_children[lpc].respawn = FALSE; + stop_child(&(crm_children[lpc]), SIGTERM); + while(crm_children[lpc].command && crm_children[lpc].pid) { + int status; + pid_t pid = 0; + + pid = wait4( + crm_children[lpc].pid, &status, WNOHANG, NULL); + + if(pid == 0) { + sched_yield (); + nanosleep (&waitsleep, 0); + continue; + + } else if(pid < 0) { + ais_perror("crm_wait_dispatch: Call to wait4(%s) failed", + crm_children[lpc].name); + } - } else if(pid < 0) { - ais_perror("crm_wait_dispatch: Call to wait4(%s) failed", - crm_children[lpc].name); + ais_notice("%s (pid=%d) confirmed dead", + crm_children[lpc].name, crm_children[lpc].pid); + + /* cleanup */ + crm_children[lpc].pid = 0; + crm_children[lpc].conn = NULL; + crm_children[lpc].async_conn = NULL; + break; } - - ais_notice("%s (pid=%d) confirmed dead", - crm_children[lpc].name, crm_children[lpc].pid); - - /* cleanup */ - crm_children[lpc].pid = 0; - crm_children[lpc].conn = NULL; - crm_children[lpc].async_conn = NULL; - break; } } - + send_cluster_id(); ais_notice("Shutdown complete"); LEAVE(""); +#ifndef AIS_WHITETANK logsys_flush (); +#endif return 0; } struct member_loop_data { char *string; }; void member_loop_fn(gpointer key, gpointer value, gpointer user_data) { crm_node_t *node = value; struct member_loop_data *data = user_data; ais_debug_2("Dumping node %u", node->id); data->string = append_member(data->string, node); } char *ais_generate_membership_data(void) { int size = 0; struct member_loop_data data; size = 14 + 32; /* + int */ ais_malloc0(data.string, size); sprintf(data.string, "", membership_seq); g_hash_table_foreach(membership_list, member_loop_fn, &data); size = strlen(data.string); data.string = realloc(data.string, size + 9) ;/* 9 = + nul */ sprintf(data.string + size, ""); return data.string; } void ais_node_list_query(void *conn, void *msg) { char *data = ais_generate_membership_data(); void *async_conn = openais_conn_partner_get(conn); - ais_debug_4("members: %s", data); + + /* send the ACK before we send any other messages */ + send_ipc_ack(conn, 1); + if(async_conn) { send_client_msg(async_conn, crm_class_members, crm_msg_none, data); } ais_free(data); - send_ipc_ack(conn, 1); } void ais_manage_notification(void *conn, void *msg) { int lpc = 0; int enable = 0; AIS_Message *ais_msg = msg; char *data = get_ais_data(ais_msg); if(ais_str_eq("true", data)) { enable = 1; } for (; lpc < SIZEOF(crm_children); lpc++) { if(crm_children[lpc].conn == conn) { ais_info("%s node notifications for %s", enable?"Enabling":"Disabling", crm_children[lpc].name); if(enable) { crm_children[lpc].flags |= crm_flag_members; } else { crm_children[lpc].flags |= crm_flag_members; crm_children[lpc].flags ^= crm_flag_members; } break; } } send_ipc_ack(conn, 2); } void send_member_notification(void) { int lpc = 0; char *update = ais_generate_membership_data(); for (; lpc < SIZEOF(crm_children); lpc++) { if(crm_children[lpc].flags & crm_flag_members) { if(crm_children[lpc].async_conn == NULL) { continue; } ais_info("Sending membership update %llu to %s", membership_seq, crm_children[lpc].name); send_client_msg(crm_children[lpc].async_conn, crm_class_members, crm_msg_none, update); } } ais_free(update); } static gboolean check_message_sanity(AIS_Message *msg, char *data) { gboolean sane = TRUE; gboolean repaired = FALSE; int dest = msg->host.type; int tmp_size = msg->header.size - sizeof(AIS_Message); if(sane && msg->header.size == 0) { ais_warn("Message with no size"); sane = FALSE; } if(sane && msg->header.error != 0) { ais_warn("Message header contains an error: %d", msg->header.error); sane = FALSE; } if(sane && ais_data_len(msg) != tmp_size) { int cur_size = ais_data_len(msg); repaired = TRUE; if(msg->is_compressed) { msg->compressed_size = tmp_size; } else { msg->size = tmp_size; } ais_warn("Repaired message payload size %d -> %d", cur_size, tmp_size); } if(sane && ais_data_len(msg) == 0) { ais_warn("Message with no payload"); sane = FALSE; } if(sane && data && msg->is_compressed == FALSE) { int str_size = strlen(data) + 1; if(ais_data_len(msg) != str_size) { int lpc = 0; ais_warn("Message payload is corrupted: expected %d bytes, got %d", ais_data_len(msg), str_size); sane = FALSE; for(lpc = (str_size - 10); lpc < msg->size; lpc++) { if(lpc < 0) { lpc = 0; } ais_debug_2("bad_data[%d]: %d / '%c'", lpc, data[lpc], data[lpc]); } } } if(sane == FALSE) { ais_err("Invalid message %d: (dest=%s:%s, from=%s:%s.%d, compressed=%d, size=%d, total=%d)", msg->id, ais_dest(&(msg->host)), msg_type2text(dest), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->is_compressed, ais_data_len(msg), msg->header.size); } else if(repaired) { ais_err("Repaired message %d: (dest=%s:%s, from=%s:%s.%d, compressed=%d, size=%d, total=%d)", msg->id, ais_dest(&(msg->host)), msg_type2text(dest), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->is_compressed, ais_data_len(msg), msg->header.size); } else { ais_debug_3("Verified message %d: (dest=%s:%s, from=%s:%s.%d, compressed=%d, size=%d, total=%d)", msg->id, ais_dest(&(msg->host)), msg_type2text(dest), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->is_compressed, ais_data_len(msg), msg->header.size); } return sane; } gboolean route_ais_message(AIS_Message *msg, gboolean local_origin) { int rc = 0; int level = LOG_WARNING; int dest = msg->host.type; ais_debug_3("Msg[%d] (dest=%s:%s, from=%s:%s.%d, remote=%s, size=%d)", msg->id, ais_dest(&(msg->host)), msg_type2text(dest), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, local_origin?"false":"true", ais_data_len(msg)); if(local_origin == FALSE) { if(msg->host.size == 0 || ais_str_eq(local_uname, msg->host.uname)) { msg->host.local = TRUE; } } if(check_message_sanity(msg, msg->data) == FALSE) { /* Dont send this message to anyone */ return FALSE; } if(msg->host.local) { void *conn = NULL; const char *lookup = NULL; if(dest == crm_msg_ais) { process_ais_message(msg); return TRUE; } else if(dest == crm_msg_lrmd) { /* lrmd messages are routed via the crm */ dest = crm_msg_crmd; } else if(dest == crm_msg_te) { /* te messages are routed via the crm - for now */ dest = crm_msg_crmd; } if(in_shutdown) { level = LOG_INFO; } AIS_CHECK(dest > 0 && dest < SIZEOF(crm_children), ais_err("Invalid destination: %d", dest); log_ais_message(LOG_ERR, msg); return FALSE; ); rc = 1; lookup = msg_type2text(dest); conn = crm_children[dest].async_conn; /* the cluster fails in weird and wonderfully obscure ways when this is not true */ AIS_ASSERT(ais_str_eq(lookup, crm_children[dest].name)); if (conn == NULL) { do_ais_log(level, "No connection to %s", crm_children[dest].name); } else if (!libais_connection_active(conn)) { do_ais_log(level, "Connection to %s is no longer active", crm_children[dest].name); crm_children[dest].async_conn = NULL; /* } else if ((queue->size - 1) == queue->used) { */ /* ais_err("Connection is throttled: %d", queue->size); */ } else { level = LOG_ERR; ais_debug_3("Delivering locally to %s (size=%d)", crm_children[dest].name, msg->header.size); +#ifdef AIS_WHITETANK + rc = openais_dispatch_send(conn, msg, msg->header.size); +#else rc = openais_conn_send_response(conn, msg, msg->header.size); +#endif } } else if(local_origin) { /* forward to other hosts */ ais_debug_3("Forwarding to cluster"); rc = send_cluster_msg_raw(msg); } else { ais_debug_3("Ignoring..."); } if(rc != 0) { do_ais_log(level, "Sending message to %s.%s failed (rc=%d)", ais_dest(&(msg->host)), msg_type2text(dest), rc); log_ais_message(level, msg); return FALSE; } return TRUE; } int send_cluster_msg_raw(AIS_Message *ais_msg) { int rc = 0; struct iovec iovec; static uint32_t msg_id = 0; AIS_Message *bz2_msg = NULL; ENTER(""); AIS_ASSERT(local_nodeid != 0); if(ais_msg->header.size != (sizeof(AIS_Message) + ais_data_len(ais_msg))) { ais_err("Repairing size mismatch: %u + %d = %d", (unsigned int)sizeof(AIS_Message), ais_data_len(ais_msg), ais_msg->header.size); ais_msg->header.size = sizeof(AIS_Message) + ais_data_len(ais_msg); } if(ais_msg->id == 0) { msg_id++; AIS_CHECK(msg_id != 0 /* detect wrap-around */, msg_id++; ais_err("Message ID wrapped around")); ais_msg->id = msg_id; } ais_msg->header.id = SERVICE_ID_MAKE(CRM_SERVICE, 0); ais_msg->sender.id = local_nodeid; ais_msg->sender.size = local_uname_len; memset(ais_msg->sender.uname, 0, MAX_NAME); memcpy(ais_msg->sender.uname, local_uname, ais_msg->sender.size); iovec.iov_base = (char *)ais_msg; iovec.iov_len = ais_msg->header.size; #if 0 if(ais_msg->is_compressed == FALSE && ais_msg->size > 1024) { char *compressed = NULL; unsigned int len = (ais_msg->size * 1.1) + 600; /* recomended size */ ais_debug_2("Creating compressed message"); ais_malloc0(compressed, len); rc = BZ2_bzBuffToBuffCompress( compressed, &len, ais_msg->data, ais_msg->size, 3, 0, 30); if(rc != BZ_OK) { ais_err("Compression failed: %d", rc); ais_free(compressed); goto send; } ais_malloc0(bz2_msg, sizeof(AIS_Message) + len + 1); memcpy(bz2_msg, ais_msg, sizeof(AIS_Message)); memcpy(bz2_msg->data, compressed, len); ais_free(compressed); bz2_msg->is_compressed = TRUE; bz2_msg->compressed_size = len; bz2_msg->header.size = sizeof(AIS_Message) + ais_data_len(bz2_msg); ais_debug("Compression details: %d -> %d", bz2_msg->size, ais_data_len(bz2_msg)); iovec.iov_base = (char *)bz2_msg; iovec.iov_len = bz2_msg->header.size; } send: #endif ais_debug_3("Sending message (size=%u)", (unsigned int)iovec.iov_len); rc = totempg_groups_mcast_joined ( openais_group_handle, &iovec, 1, TOTEMPG_SAFE); if(rc == 0 && ais_msg->is_compressed == FALSE) { ais_debug_2("Message sent: %.80s", ais_msg->data); } AIS_CHECK(rc == 0, ais_err("Message not sent (%d)", rc)); ais_free(bz2_msg); LEAVE(""); return rc; } #define min(x,y) (x)<(y)?(x):(y) void send_cluster_id(void) { int rc = 0; int lpc = 0; int len = 0; struct iovec iovec; struct crm_identify_msg_s *msg = NULL; ENTER(""); AIS_ASSERT(local_nodeid != 0); ais_malloc0(msg, sizeof(struct crm_identify_msg_s)); msg->header.size = sizeof(struct crm_identify_msg_s); msg->id = local_nodeid; msg->header.id = SERVICE_ID_MAKE(CRM_SERVICE, 1); len = min(local_uname_len, MAX_NAME-1); memset(msg->uname, 0, MAX_NAME); memcpy(msg->uname, local_uname, len); len = min(strlen(VERSION), MAX_NAME-1); memset(msg->version, 0, MAX_NAME); memcpy(msg->version, VERSION, len); msg->votes = 1; msg->pid = getpid(); msg->processes = crm_proc_ais; for (lpc = 0; lpc < SIZEOF(crm_children); lpc++) { if(crm_children[lpc].pid != 0) { msg->processes |= crm_children[lpc].flag; } } ais_debug("Local update: %u", local_nodeid); update_member( local_nodeid, membership_seq, msg->votes, msg->processes, NULL, NULL); iovec.iov_base = (char *)msg; iovec.iov_len = msg->header.size; rc = totempg_groups_mcast_joined ( openais_group_handle, &iovec, 1, TOTEMPG_SAFE); AIS_CHECK(rc == 0, ais_err("Message not sent (%d)", rc)); ais_free(msg); LEAVE(""); } diff --git a/crm/ais/utils.c b/crm/ais/utils.c index dd6bfaf8b0..cc5cdb4c34 100644 --- a/crm/ais/utils.c +++ b/crm/ais/utils.c @@ -1,481 +1,484 @@ /* * 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 #include "./utils.h" int in_shutdown = FALSE; extern int send_cluster_msg_raw(AIS_Message *ais_msg); void log_ais_message(int level, AIS_Message *msg) { char *data = get_ais_data(msg); do_ais_log(level, "Msg[%d] (dest=%s:%s, from=%s:%s.%d, remote=%s, size=%d): %.90s", msg->id, ais_dest(&(msg->host)), msg_type2text(msg->host.type), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->sender.uname==local_uname?"false":"true", ais_data_len(msg), data); ais_free(data); } gboolean process_ais_message(AIS_Message *msg) { char *data = get_ais_data(msg); do_ais_log(LOG_NOTICE, "Msg[%d] (dest=%s:%s, from=%s:%s.%d, remote=%s, size=%d): %.90s", msg->id, ais_dest(&(msg->host)), msg_type2text(msg->host.type), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->sender.uname==local_uname?"false":"true", ais_data_len(msg), data); ais_free(data); return TRUE; } static int ais_string_to_boolean(const char * s) { int rc = 0; if(s == NULL) { return rc; } if(strcasecmp(s, "true") == 0 || strcasecmp(s, "on") == 0 || strcasecmp(s, "yes") == 0 || strcasecmp(s, "y") == 0 || strcasecmp(s, "1") == 0) { rc = 1; } return rc; } gboolean spawn_child(crm_child_t *child) { int rc = 0; int lpc = 0; struct rlimit oflimits; const char *devnull = "/dev/null"; const char *use_valgrind = getenv("HA_VALGRIND_ENABLED"); if(child->command == NULL) { ais_info("Nothing to do for child \"%s\"", child->name); return TRUE; } child->pid = fork(); AIS_ASSERT(child->pid != -1); if(child->pid > 0) { /* parent */ ais_info("Forked child %d for process %s", child->pid, child->name); return TRUE; } /* Child */ ais_debug("Executing \"%s (%s)\" (pid %d)", child->command, child->name, (int) getpid()); if(child->uid > 0) { rc = setuid(child->uid); if(rc < 0) { ais_perror("Could not set user to %d", child->uid); } } /* A precautionary measure */ getrlimit(RLIMIT_NOFILE, &oflimits); for (; lpc < oflimits.rlim_cur; lpc++) { close(lpc); } (void)open(devnull, O_RDONLY); /* Stdin: fd 0 */ (void)open(devnull, O_WRONLY); /* Stdout: fd 1 */ (void)open(devnull, O_WRONLY); /* Stderr: fd 2 */ if(ais_string_to_boolean(use_valgrind)) { char *opts[] = { ais_strdup(VALGRIND_BIN), ais_strdup(child->command), NULL }; (void)execvp(VALGRIND_BIN, opts); } else { char *opts[] = { ais_strdup(child->command), NULL }; (void)execvp(child->command, opts); } ais_perror("FATAL: Cannot exec %s", child->command); exit(100); return TRUE; /* never reached */ } gboolean stop_child(crm_child_t *child, int signal) { if(signal == 0) { signal = SIGTERM; } if(child->command == NULL) { ais_info("Nothing to do for child \"%s\"", child->name); return TRUE; } ais_debug("Stopping CRM child \"%s\"", child->name); if (child->pid <= 0) { ais_debug_2("Client %s not running", child->name); return TRUE; } errno = 0; if(kill(child->pid, signal) == 0) { ais_notice("Sent -%d to %s: [%d]", signal, child->name, child->pid); } else { ais_perror("Sent -%d to %s: [%d]", signal, child->name, child->pid); } return TRUE; } void destroy_ais_node(gpointer data) { crm_node_t *node = data; ais_info("Destroying entry for node %u", node->id); ais_free(node->addr); ais_free(node->uname); ais_free(node->state); ais_free(node); } int update_member(unsigned int id, unsigned long long seq, int32_t votes, uint32_t procs, const char *uname, const char *state) { int changed = 0; crm_node_t *node = NULL; node = g_hash_table_lookup(membership_list, GUINT_TO_POINTER(id)); if(node == NULL) { ais_malloc0(node, sizeof(crm_node_t)); ais_info("Creating entry for node %u born on %llu", id, seq); node->id = id; node->addr = NULL; node->state = ais_strdup("unknown"); g_hash_table_insert(membership_list, GUINT_TO_POINTER(id), node); node = g_hash_table_lookup(membership_list, GUINT_TO_POINTER(id)); } if(seq != 0) { node->last_seen = seq; } if(uname != NULL) { if(node->uname == NULL || ais_str_eq(node->uname, uname) == FALSE) { ais_info("%p Node %u now known as %s (was: %s)", node, id, uname, node->uname); ais_free(node->uname); node->uname = ais_strdup(uname); changed = TRUE; } } if(procs != 0 && procs != node->processes) { ais_info("Node %s now has process list: %.32x (%u)", node->uname, procs, procs); node->processes = procs; changed = TRUE; } if(votes >= 0 && votes != node->votes) { ais_info("Node %s now has %d quorum votes (was %d)", node->uname, votes, node->votes); node->votes = votes; changed = TRUE; } if(state != NULL) { if(node->state == NULL || ais_str_eq(node->state, state) == FALSE) { ais_free(node->state); node->state = ais_strdup(state); ais_info("Node %u/%s is now: %s", id, node->uname?node->uname:"unknown", state); changed = TRUE; } } AIS_ASSERT(node != NULL); return changed; } void delete_member(uint32_t id, const char *uname) { if(uname == NULL) { g_hash_table_remove(membership_list, GUINT_TO_POINTER(id)); return; } ais_err("Deleting by uname is not yet supported"); } const char *member_uname(uint32_t id) { crm_node_t *node = g_hash_table_lookup( membership_list, GUINT_TO_POINTER(id)); if(node == NULL) { return ".unknown."; } if(node->uname == NULL) { return ".pending."; } return node->uname; } #define MEMBER_FORMAT "" char *append_member(char *data, crm_node_t *node) { int size = 1; /* nul */ int offset = 0; if(node->uname == NULL) { return data; } if(data) { size = strlen(data); } offset = size; size += strlen(MEMBER_FORMAT); size += 32; /* node->id */ size += strlen(node->uname); size += strlen(node->state); data = realloc(data, size); if(node->addr) { size += strlen(node->addr); } sprintf(data+offset, MEMBER_FORMAT, node->id, node->uname, node->state, membership_seq, node->votes, node->processes, node->addr?node->addr:""); return data; } void swap_sender(AIS_Message *msg) { int tmp = 0; char tmp_s[256]; tmp = msg->host.type; msg->host.type = msg->sender.type; msg->sender.type = tmp; tmp = msg->host.type; msg->host.size = msg->sender.type; msg->sender.type = tmp; memcpy(tmp_s, msg->host.uname, 256); memcpy(msg->host.uname, msg->sender.uname, 256); memcpy(msg->sender.uname, tmp_s, 256); } char *get_ais_data(AIS_Message *msg) { int rc = BZ_OK; char *uncompressed = NULL; unsigned int new_size = msg->size; if(msg->is_compressed == FALSE) { uncompressed = strdup(msg->data); } else { ais_malloc0(uncompressed, new_size); rc = BZ2_bzBuffToBuffDecompress( uncompressed, &new_size, msg->data, msg->compressed_size, 1, 0); if(rc != BZ_OK) { ais_info("rc=%d, new=%u expected=%u", rc, new_size, msg->size); } AIS_ASSERT(rc == BZ_OK); AIS_ASSERT(new_size == msg->size); } return uncompressed; } int send_cluster_msg( enum crm_ais_msg_types type, const char *host, const char *data) { int rc = 0; int data_len = 0; AIS_Message *ais_msg = NULL; int total_size = sizeof(AIS_Message); ENTER(""); AIS_ASSERT(local_nodeid != 0); if(data != NULL) { data_len = 1 + strlen(data); total_size += data_len; } ais_malloc0(ais_msg, total_size); ais_msg->header.size = total_size; ais_msg->header.id = 0; ais_msg->size = data_len; memcpy(ais_msg->data, data, data_len); ais_msg->sender.type = crm_msg_ais; ais_msg->host.type = type; ais_msg->host.id = 0; if(host) { ais_msg->host.size = strlen(host); memset(ais_msg->host.uname, 0, MAX_NAME); memcpy(ais_msg->host.uname, host, ais_msg->host.size); /* ais_msg->host.id = nodeid_lookup(host); */ } else { ais_msg->host.type = type; ais_msg->host.size = 0; memset(ais_msg->host.uname, 0, MAX_NAME); } rc = send_cluster_msg_raw(ais_msg); ais_free(ais_msg); LEAVE(""); return rc; } int send_client_msg( void *conn, enum crm_ais_msg_class class, enum crm_ais_msg_types type, const char *data) { int rc = 0; int data_len = 0; int total_size = sizeof(AIS_Message); AIS_Message *ais_msg = NULL; static int msg_id = 0; ENTER(""); AIS_ASSERT(local_nodeid != 0); msg_id++; AIS_ASSERT(msg_id != 0 /* wrap-around */); if(data != NULL) { data_len = 1 + strlen(data); } total_size += data_len; ais_malloc0(ais_msg, total_size); ais_msg->id = msg_id; ais_msg->header.size = total_size; ais_msg->header.id = class; ais_msg->size = data_len; memcpy(ais_msg->data, data, data_len); ais_msg->host.type = type; ais_msg->host.size = 0; memset(ais_msg->host.uname, 0, MAX_NAME); ais_msg->host.id = 0; ais_msg->sender.type = crm_msg_ais; ais_msg->sender.size = local_uname_len; memset(ais_msg->sender.uname, 0, MAX_NAME); memcpy(ais_msg->sender.uname, local_uname, ais_msg->sender.size); ais_msg->sender.id = local_nodeid; rc = 1; if (conn == NULL) { ais_err("No connection"); } else if (!libais_connection_active(conn)) { ais_warn("Connection no longer active"); /* } else if ((queue->size - 1) == queue->used) { */ /* ais_err("Connection is throttled: %d", queue->size); */ } else { +#ifndef AIS_WHITETANK rc = openais_conn_send_response (conn, ais_msg, total_size); +#else + rc = openais_dispatch_send (conn, ais_msg, total_size); +#endif AIS_CHECK(rc == 0, ais_err("Message not sent (%d): %s", rc, data?data:"")); } - ais_debug_5("Sent %d:%s", class, data); ais_free(ais_msg); LEAVE(""); return rc; } int objdb_get_string( struct objdb_iface_ver0 *objdb, unsigned int object_service_handle, char *key, char **value, const char *fallback) { *value = NULL; if(object_service_handle > 0) { objdb->object_key_get( object_service_handle, key, strlen(key), (void**)value, NULL); } if (*value) { ais_info("Found '%s' for option %s", *value, key); return 0; } ais_info("Defaulting to '%s' for option %s", fallback, key); *value = ais_strdup(fallback); return -1; } int objdb_get_int( struct objdb_iface_ver0 *objdb, unsigned int object_service_handle, char *key, unsigned int *int_value, const char *fallback) { char *value = NULL; objdb_get_string(objdb, object_service_handle, key, &value, fallback); if (value) { *int_value = atoi(value); return 0; } return -1; } diff --git a/crm/ais/utils.h b/crm/ais/utils.h index bf34e44ce6..540fbd5bf6 100644 --- a/crm/ais/utils.h +++ b/crm/ais/utils.h @@ -1,180 +1,192 @@ /* * 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 AIS_CRM_UTILS__H #define AIS_CRM_UTILS__H +#include +#include +#ifdef AIS_WHITETANK +# include +# define openais_conn_partner_get(conn) conn +extern int openais_response_send (void *conn, void *msg, int mlen); +extern int openais_dispatch_send (void *conn, void *msg, int mlen); +static int libais_connection_active (void *conn) { return 1; } + +#else +# include +LOGSYS_DECLARE_SUBSYS("crm", LOG_LEVEL_DEBUG); + /* from openais/exec/ipc.h */ extern int openais_conn_send_response (void *conn, void *msg, int mlen); extern int libais_connection_active (void *conn); +#endif -#include -#include -LOGSYS_DECLARE_SUBSYS("crm", LOG_LEVEL_DEBUG); /* #include "plugin.h" */ #define SIZEOF(a) (sizeof(a) / sizeof(a[0])) #define CRM_MESSAGE_IPC_ACK 1 #define CRM_SERVICE 16 typedef struct crm_child_s { int pid; long flag; long flags; + int start_seq; int respawn_count; gboolean respawn; const char *name; int uid; const char *command; void *conn; void *async_conn; } crm_child_t; extern void destroy_ais_node(gpointer data); extern void delete_member(uint32_t id, const char *uname); extern int update_member(unsigned int id, unsigned long long seq, int32_t votes, uint32_t procs, const char *uname, const char *state); extern const char *member_uname(uint32_t id); extern char *append_member(char *data, crm_node_t *node); extern void member_loop_fn(gpointer key, gpointer value, gpointer user_data); extern gboolean stop_child(crm_child_t *child, int signal); extern gboolean spawn_child(crm_child_t *child); extern void swap_sender(AIS_Message *msg); extern char *get_ais_data(AIS_Message *msg); extern gboolean route_ais_message(AIS_Message *msg, gboolean local); extern gboolean process_ais_message(AIS_Message *msg); extern int send_cluster_msg( enum crm_ais_msg_types type, const char *host, const char *data); extern int send_client_msg(void *conn, enum crm_ais_msg_class class, enum crm_ais_msg_types type, const char *data); extern void send_member_notification(void); extern void log_ais_message(int level, AIS_Message *msg); extern int objdb_get_int( struct objdb_iface_ver0 *objdb, unsigned int object_service_handle, char *key, unsigned int *int_value, const char *fallback); extern int objdb_get_string( struct objdb_iface_ver0 *objdb, unsigned int object_service_handle, char *key, char **value, const char *fallback); extern GHashTable *membership_list; extern pthread_t crm_wait_thread; extern int plugin_log_level; extern char *local_uname; extern int local_uname_len; extern unsigned int local_nodeid; extern unsigned long long membership_seq; extern int in_shutdown; static inline const char *level2char(int level) { switch(level) { case LOG_CRIT: return "CRIT"; case LOG_ERR: return "ERROR"; case LOG_WARNING: return "WARN"; case LOG_NOTICE: return "notice"; case LOG_INFO: return "info"; } return "debug"; } #define do_ais_log(level, fmt, args...) do { \ if(plugin_log_level < (level)) { \ continue; \ } else if((level) > LOG_DEBUG) { \ log_printf(LOG_DEBUG, "debug%d: %s: " fmt, \ level-LOG_INFO, __PRETTY_FUNCTION__ , ##args); \ } else { \ log_printf(level, "%s: %s: " fmt, level2char(level), \ __PRETTY_FUNCTION__ , ##args); \ } \ } while(0) #define ais_perror(fmt, args...) log_printf( \ LOG_ERR, "%s: " fmt ": (%d) %s", \ __PRETTY_FUNCTION__ , ##args, errno, strerror(errno)) #define ais_crit(fmt, args...) do_ais_log(LOG_CRIT, fmt , ##args) #define ais_err(fmt, args...) do_ais_log(LOG_ERR, fmt , ##args) #define ais_warn(fmt, args...) do_ais_log(LOG_WARNING, fmt , ##args) #define ais_notice(fmt, args...) do_ais_log(LOG_NOTICE, fmt , ##args) #define ais_info(fmt, args...) do_ais_log(LOG_INFO, fmt , ##args) #define ais_debug(fmt, args...) do_ais_log(LOG_DEBUG, fmt , ##args) #define ais_debug_2(fmt, args...) do_ais_log(LOG_DEBUG+1, fmt , ##args) #define ais_debug_3(fmt, args...) do_ais_log(LOG_DEBUG+2, fmt , ##args) #define ais_debug_4(fmt, args...) do_ais_log(LOG_DEBUG+3, fmt , ##args) #define ais_debug_5(fmt, args...) do_ais_log(LOG_DEBUG+4, fmt , ##args) #define ais_debug_6(fmt, args...) do_ais_log(LOG_DEBUG+5, fmt , ##args) #define ais_malloc0(malloc_obj, length) do { \ malloc_obj = malloc(length); \ if(malloc_obj == NULL) { \ abort(); \ } \ memset(malloc_obj, 0, length); \ } while(0) #define ais_free(obj) do { \ if(obj) { \ free(obj); \ obj = NULL; \ } \ } while(0) #define AIS_ASSERT(expr) if((expr) == FALSE) { \ ais_crit("Assertion failure line %d: %s", __LINE__, #expr); \ abort(); \ } #define AIS_CHECK(expr, failure_action) if((expr) == FALSE) { \ ais_err("Non fatal assertion failure line %d: %s", __LINE__, #expr); \ failure_action; \ } static inline char *ais_strdup(const char *src) { char *dup = NULL; if(src == NULL) { return NULL; } ais_malloc0(dup, strlen(src) + 1); return strcpy(dup, src); } static inline gboolean ais_str_eq(const char *a, const char *b) { if(a == NULL || b == NULL) { return FALSE; } else if(a == b) { return TRUE; } else if(strcasecmp(a, b) == 0) { return TRUE; } return FALSE; } #endif diff --git a/cts/CTSlab.py.in b/cts/CTSlab.py.in index 9757d933e5..ea982b28f4 100755 --- a/cts/CTSlab.py.in +++ b/cts/CTSlab.py.in @@ -1,841 +1,847 @@ #!@PYTHON@ '''CTS: Cluster Testing System: Lab environment module ''' __copyright__=''' Copyright (C) 2001,2005 Alan Robertson Licensed under the GNU GPL. ''' # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. from UserDict import UserDict import sys, time, types, string, syslog, random, os, string, signal, traceback from CTS import ClusterManager from CM_hb import HeartbeatCM from CTStests import BSC_AddResource from socket import gethostbyname_ex tests = None cm = None old_handler = None DefaultFacility = "daemon" def sig_handler(signum, frame) : if cm != None: cm.log("Interrupted by signal %d"%signum) if signum == 10 and tests != None : tests.summarize() if signum == 15 : sys.exit(1) class ResetMechanism: def reset(self, node): raise ValueError("Abstract class member (reset)") class Stonith(ResetMechanism): def __init__(self, sttype="external/ssh", pName=None, pValue=None , path="@sbindir@/stonith"): self.pathname=path self.configName=pName self.configValue=pValue self.stonithtype=sttype def reset(self, node): if self.configValue == None : config=node else: config=self.configValue cmdstring = "%s -t '%s' -p '%s' '%s' 2>/dev/null" % (self.pathname , self.stonithtype, config, node) return (os.system(cmdstring) == 0) class Stonithd(ResetMechanism): def __init__(self, nodes, sttype = 'external/ssh'): self.sttype = sttype self.nodes = nodes self.query_cmd_pat = '@libdir@/heartbeat/stonithdtest/apitest 0 %s 20000 0' self.reset_cmd_pat = '@libdir@/heartbeat/stonithdtest/apitest 1 %s 20000 0' self.poweron_cmd_pat = '@libdir@/heartbeat/stonithdtest/apitest 2 %s 20000 0' self.poweroff_cmd_pat= '@libdir@/heartbeat/stonithdtest/apitest 3 %s 20000 0' self.lrmd_add_pat = '@libdir@/heartbeat/lrmadmin -A %s stonith ' + sttype + ' NULL hostlist=%s' self.lrmd_start_pat = '@libdir@/heartbeat/lrmadmin -E %s start 0 0 EVERYTIME' self.lrmd_stop_pat = '@libdir@/heartbeat/lrmadmin -E %s stop 0 0 EVERYTIME' self.lrmd_del_pat = '@libdir@/heartbeat/lrmadmin -D %s' self.rsc_id = 'my_stonithd_id' self.command = "@SSH@ -l root -n -x" self.command_noblock = "@SSH@ -f -l root -n -x" self.stonithd_started_nodes = [] self.fail_reason = '' def _remote_exec(self, node, cmnd): return (os.system("%s %s %s > /dev/null" % (self.command, node, cmnd)) == 0) def _remote_readlines(self, node, cmnd): f = os.popen("%s %s %s" % (self.command, node, cmnd)) return f.readlines() def _stonithd_started(self, node): return node in self.stonithd_started_nodes def _start_stonithd(self, node, hosts): hostlist = string.join(hosts, ',') lrmd_add_cmd = self.lrmd_add_pat % (self.rsc_id, hostlist) ret = self._remote_exec(node, lrmd_add_cmd) if not ret:return ret lrmd_start_cmd = self.lrmd_start_pat % self.rsc_id ret = self._remote_exec(node, lrmd_start_cmd) if not ret:return ret self.stonithd_started_nodes.append(node) return 1 def _stop_stonithd(self, node): lrmd_stop_cmd = self.lrmd_stop_pat % self.rsc_id ret = self._remote_exec(node, lrmd_stop_cmd) if not ret:return ret lrmd_del_cmd = self.lrmd_del_pat % self.rsc_id ret = self._remote_exec(node, lrmd_del_cmd) if not ret:return ret self.stonithd_started_nodes.remove(node) return 1 def _do_stonith(self, init_node, target_node, action): stonithd_started = self._stonithd_started(init_node) if not stonithd_started: ret = self._start_stonithd(init_node, [target_node]) if not ret: self.fail_reason = "failed to start stonithd on node %s" % init_node return ret command = "" if action == "RESET": command = self.reset_cmd_pat % target_node elif action == "POWEROFF": command = self.poweroff_cmd_pat % target_node elif action == "POWERON": command = self.poweron_cmd_pat % target_node else: self.fail_reason = "unknown opration type %s" % action return 0 lines = self._remote_readlines(init_node, command) result = "".join(lines) if not stonithd_started: self._stop_stonithd(init_node) index = result.find("result=0") if index == -1: self.fail_reason = "unexpected stonithd status: %s" % result return 0 return 1 # Should we randomly choose a node as init_node here if init_node not specified? def reset(self, init_node, target_node): return self._do_stonith(init_node, target_node, "RESET") def poweron(self, init_node, target_node): return self._do_stonith(init_node, target_node, "POWERON") def poweroff(self, init_node, target_node): return self._do_stonith(init_node, target_node, "POWEROFF") class Logger: TimeFormat = "%b %d %H:%M:%S\t" def __call__(self, lines): raise ValueError("Abstract class member (__call__)") def write(self, line): return self(line.rstrip()) def writelines(self, lines): for s in lines: self.write(s) return 1 def flush(self): return 1 def isatty(self): return None class SysLog(Logger): # http://docs.python.org/lib/module-syslog.html defaultsource="CTS" map = { "kernel": syslog.LOG_KERN, "user": syslog.LOG_USER, "mail": syslog.LOG_MAIL, "daemon": syslog.LOG_DAEMON, "auth": syslog.LOG_AUTH, "lpr": syslog.LOG_LPR, "news": syslog.LOG_NEWS, "uucp": syslog.LOG_UUCP, "cron": syslog.LOG_CRON, "local0": syslog.LOG_LOCAL0, "local1": syslog.LOG_LOCAL1, "local2": syslog.LOG_LOCAL2, "local3": syslog.LOG_LOCAL3, "local4": syslog.LOG_LOCAL4, "local5": syslog.LOG_LOCAL5, "local6": syslog.LOG_LOCAL6, "local7": syslog.LOG_LOCAL7, } def __init__(self, labinfo): if labinfo.has_key("syslogsource"): self.source=labinfo["syslogsource"] else: self.source=SysLog.defaultsource if labinfo.has_key("SyslogFacility"): self.facility=labinfo["SyslogFacility"] else: self.facility=DefaultFacility if SysLog.map.has_key(self.facility): self.facility=SysLog.map[self.facility] syslog.openlog(self.source, 0, self.facility) def setfacility(self, facility): self.facility = facility if SysLog.map.has_key(self.facility): self.facility=SysLog.map[self.facility] syslog.closelog() syslog.openlog(self.source, 0, self.facility) def __call__(self, lines): if isinstance(lines, types.StringType): syslog.syslog(lines) else: for line in lines: syslog.syslog(line) def name(self): return "Syslog" class StdErrLog(Logger): def __init__(self, labinfo): pass def __call__(self, lines): t = time.strftime(Logger.TimeFormat, time.localtime(time.time())) if isinstance(lines, types.StringType): sys.__stderr__.writelines([t, lines, "\n"]) else: for line in lines: sys.__stderr__.writelines([t, line, "\n"]) sys.__stderr__.flush() def name(self): return "StdErrLog" class FileLog(Logger): def __init__(self, labinfo, filename=None): if filename == None: filename=labinfo["LogFileName"] self.logfile=filename import os self.hostname = os.uname()[1]+" " self.source = "CTS: " def __call__(self, lines): fd = open(self.logfile, "a") t = time.strftime(Logger.TimeFormat, time.localtime(time.time())) if isinstance(lines, types.StringType): fd.writelines([t, self.hostname, self.source, lines, "\n"]) else: for line in lines: fd.writelines([t, self.hostname, self.source, line, "\n"]) fd.close() def name(self): return "FileLog" class CtsLab(UserDict): '''This class defines the Lab Environment for the Cluster Test System. It defines those things which are expected to change from test environment to test environment for the same cluster manager. It is where you define the set of nodes that are in your test lab what kind of reset mechanism you use, etc. This class is derived from a UserDict because we hold many different parameters of different kinds, and this provides provide a uniform and extensible interface useful for any kind of communication between the user/administrator/tester and CTS. At this point in time, it is the intent of this class to model static configuration and/or environmental data about the environment which doesn't change as the tests proceed. Well-known names (keys) are an important concept in this class. The HasMinimalKeys member function knows the minimal set of well-known names for the class. The following names are standard (well-known) at this time: nodes An array of the nodes in the cluster reset A ResetMechanism object logger An array of objects that log strings... CMclass The type of ClusterManager we are running (This is a class object, not a class instance) RandSeed Random seed. It is a triple of bytes. (optional) HAdir Base directory for HA installation The CTS code ignores names it doesn't know about/need. The individual tests have access to this information, and it is perfectly acceptable to provide hints, tweaks, fine-tuning directions or other information to the tests through this mechanism. ''' def __init__(self, nodes): self.data = {} self["nodes"] = nodes self.MinimalKeys=["nodes", "reset", "logger", "CMclass", "HAdir"] def HasMinimalKeys(self): 'Return TRUE if our object has the minimal set of keys/values in it' result = 1 for key in self.MinimalKeys: if not self.has_key(key): result = None return result def SupplyDefaults(self): if not self.has_key("logger"): self["logger"] = (SysLog(self), StdErrLog(self)) if not self.has_key("reset"): self["reset"] = Stonith() if not self.has_key("CMclass"): self["CMclass"] = HeartbeatCM if not self.has_key("LogFileName"): self["LogFileName"] = "@HA_VARLOGDIR@/ha-log" if not self.has_key("logrestartcmd"): self["logrestartcmd"] = "@INITDIR@/syslog restart" if not self.has_key("logfacility"): LogFacility = DefaultFacility # # Now set up our random number generator... # self.RandomGen = random.Random() # Get a random seed for the random number generator. if self.has_key("RandSeed"): randseed = self["RandSeed"] self.log("Random seed is: " + str(randseed)) self.RandomGen.seed(str(randseed)) else: randseed = int(time.time()) self.log("Random seed is: " + str(randseed)) self.RandomGen.seed(str(randseed)) def log(self, args): "Log using each of the supplied logging methods" for logfcn in self._logfunctions: logfcn(string.strip(args)) def debug(self, args): "Log using each of the supplied logging methods" for logfcn in self._logfunctions: if logfcn.name() != "StdErrLog": logfcn("debug: %s" % string.strip(args)) def __setitem__(self, key, value): '''Since this function gets called whenever we modify the dictionary (object), we can (and do) validate those keys that we know how to validate. For the most part, we know how to validate the "MinimalKeys" elements. ''' # # List of nodes in the system # if key == "nodes": self.Nodes = {} for node in value: # I don't think I need the IP address, etc. but this validates # the node name against /etc/hosts and/or DNS, so it's a # GoodThing(tm). try: self.Nodes[node] = gethostbyname_ex(node) except: print node+" not found in DNS... aborting" raise # # Reset Mechanism # elif key == "reset": if not issubclass(value.__class__, ResetMechanism): raise ValueError("'reset' Value must be a subclass" " of ResetMechanism") # # List of Logging Mechanism(s) # elif key == "logger": if len(value) < 1: raise ValueError("Must have at least one logging mechanism") for logger in value: if not callable(logger): raise ValueError("'logger' elements must be callable") self._logfunctions = value # # Cluster Manager Class # elif key == "CMclass": if not issubclass(value, ClusterManager): raise ValueError("'CMclass' must be a subclass of" " ClusterManager") # # Initial Random seed... # #elif key == "RandSeed": # if len(value) != 3: # raise ValueError("'Randseed' must be a 3-element list/tuple") # for elem in value: # if not isinstance(elem, types.IntType): # raise ValueError("'Randseed' list must all be ints") self.data[key] = value def IsValidNode(self, node): 'Return TRUE if the given node is valid' return self.Nodes.has_key(node) def __CheckNode(self, node): "Raise a ValueError if the given node isn't valid" if not self.IsValidNode(node): raise ValueError("Invalid node [%s] in CheckNode" % node) def RandomNode(self): '''Choose a random node from the cluster''' return self.RandomGen.choice(self["nodes"]) def ResetNode(self, node): "Reset a node, (normally) using a hardware mechanism" self.__CheckNode(node) return self["reset"].reset(node) def ResetNode2(self, init_node, target_node, reasons): self.__CheckNode(target_node) stonithd = Stonithd(self["nodes"]) ret = stonithd.reset(init_node, target_node) if not ret: reasons.append(stonithd.fail_reason) return ret def usage(arg): print "Illegal argument " + arg print "usage: " + sys.argv[0] \ + " --directory config-directory" \ + " -D config-directory" \ + " --logfile system-logfile-name" \ + " --trunc (truncate logfile before starting)" \ + " -L system-logfile-name" \ + " --limit-nodes maxnumnodes" \ + " --xmit-loss lost-rate(0.0-1.0)" \ + " --recv-loss lost-rate(0.0-1.0)" \ + " --suppressmonitoring" \ + " --syslog-facility syslog-facility" \ + " --facility syslog-facility" \ + " --choose testcase-name" \ + " --test-ip-base ip" \ + " --oprofile \"whitespace separated list of nodes to oprofile\"" \ + " (-2 |"\ + " -v2 |"\ + " --crm |"\ + " --classic)"\ + " (--populate-resources | -r)" \ + " --resource-can-stop" \ + " --stonith (1 | 0 | yes | no)" \ + " --stonith-type type" \ + " --stonith-args name=value" \ + " --standby (1 | 0 | yes | no)" \ + " --fencing (1 | 0 | yes | no)" \ + " --suppress_cib_writes (1 | 0 | yes | no)" \ + " -lstests" \ + " --seed" \ + " [number-of-iterations]" sys.exit(1) # # A little test code... # if __name__ == '__main__': from CTSaudits import AuditList from CTStests import TestList,RandomTests from CTS import Scenario, InitClusterManager, PingFest, PacketLoss, BasicSanityCheck import CM_hb HAdir = "@sysconfdir@/ha.d" LogFile = "@HA_VARLOGDIR@/ha-log-"+DefaultFacility DoStonith = 1 DoStandby = 1 DoFencing = 1 NumIter = 500 SuppressMonitoring = None Version = 1 CIBfilename = None CIBResource = 0 ClobberCIB = 0 LimitNodes = 0 TestCase = None LogFacility = None TruncateLog = 0 ResCanStop = 0 XmitLoss = "0.0" RecvLoss = "0.0" IPBase = "127.0.0.10" SuppressCib = 1 DoBSC = 0 ListTests = 0 HaveSeed = 0 oprofile = None warn_inactive = 0 StonithType = "ssh" StonithParams = None StonithParams = "hostlist=dynamic".split('=') # # The values of the rest of the parameters are now properly derived from # the configuration files. # # Stonith is configurable because it's slow, I have a few machines which # don't reboot very reliably, and it can mild damage to your machine if # you're using a real power switch. # # Standby is configurable because the test is very heartbeat specific # and I haven't written the code to set it properly yet. Patches are # being accepted... # Set the signal handler signal.signal(15, sig_handler) signal.signal(10, sig_handler) # Process arguments... skipthis=None args=sys.argv[1:] for i in range(0, len(args)): if skipthis: skipthis=None continue elif args[i] == "-D" or args[i] == "--directory": skipthis=1 HAdir = args[i+1] elif args[i] == "-l" or args[i] == "--limit-nodes": skipthis=1 LimitNodes = int(args[i+1]) elif args[i] == "-r" or args[i] == "--populate-resources": CIBResource = 1 elif args[i] == "-L" or args[i] == "--logfile": skipthis=1 LogFile = args[i+1] elif args[i] == "--test-ip-base": skipthis=1 IPBase = args[i+1] elif args[i] == "--oprofile": skipthis=1 oprofile = args[i+1].split(' ') elif args[i] == "--trunc": TruncateLog=1 elif args[i] == "-v2": Version=2 elif args[i] == "-lstests": ListTests=1 elif args[i] == "--stonith": skipthis=1 if args[i+1] == "1" or args[i+1] == "yes": DoStonith=1 elif args[i+1] == "0" or args[i+1] == "no": DoStonith=0 else: usage(args[i+1]) elif args[i] == "--stonith-type": StonithType = args[i+1] skipthis=1 elif args[i] == "--stonith-args": StonithParams = args[i+1].split('=') skipthis=1 elif args[i] == "--suppress-cib-writes": skipthis=1 if args[i+1] == "1" or args[i+1] == "yes": SuppressCib=1 elif args[i+1] == "0" or args[i+1] == "no": SuppressCib=0 else: usage(args[i+1]) elif args[i] == "--bsc": DoBSC=1 elif args[i] == "--standby": skipthis=1 if args[i+1] == "1" or args[i+1] == "yes": DoStandby=1 elif args[i+1] == "0" or args[i+1] == "no": DoStandby=0 else: usage(args[i+1]) elif args[i] == "--fencing": skipthis=1 if args[i+1] == "1" or args[i+1] == "yes": DoFencing=1 elif args[i+1] == "0" or args[i+1] == "no": DoFencing=0 else: usage(args[i+1]) elif args[i] == "--suppressmonitoring": SuppressMonitoring = 1 elif args[i] == "--resource-can-stop": ResCanStop = 1 elif args[i] == "-3" or args[i] == "--ais": Version = 3 elif args[i] == "-2" or args[i] == "--crm": Version = 2 elif args[i] == "-1" or args[i] == "--classic": Version = 1 elif args[i] == "--clobber-cib" or args[i] == "-c": ClobberCIB = 1 elif args[i] == "--cib-filename": skipthis=1 CIBfilename = args[i+1] elif args[i] == "--xmit-loss": try: float(args[i+1]) except ValueError: print ("--xmit-loss parameter should be float") usage(args[i+1]) skipthis=1 XmitLoss = args[i+1] elif args[i] == "--recv-loss": try: float(args[i+1]) except ValueError: print ("--recv-loss parameter should be float") usage(args[i+1]) skipthis=1 RecvLoss = args[i+1] elif args[i] == "--choose": skipthis=1 TestCase = args[i+1] + elif args[i] == "--nodes": + skipthis=1 + node_list = args[i+1].split(' ') elif args[i] == "--syslog-facility" or args[i] == "--facility": skipthis=1 LogFacility = args[i+1] elif args[i] == "--seed": skipthis=1 Seed=args[i+1] HaveSeed = 1 elif args[i] == "--warn-inactive": warn_inactive = 1 else: NumIter=int(args[i]) if not oprofile: oprofile = [] # # This reading of HBconfig here is ugly, and I suppose ought to # be done by the Cluster manager. This would probably mean moving the # list of cluster nodes into the ClusterManager class. A good thought # for our Copious Spare Time in the future... # config = CM_hb.HBConfig(HAdir) - node_list = config.Parameters["node"] + if not node_list: + node_list = config.Parameters["node"] + else: + config.Parameters["node"] = node_list if DoBSC: NumIter = 2 Version = 2 while len(node_list) > 1: node_list.pop(len(node_list)-1) if LogFacility == None: if config.Parameters.has_key("logfacility"): LogFacility = config.Parameters["logfacility"][0] else: LogFacility = DefaultFacility if LimitNodes > 0: if len(node_list) > LimitNodes: print("Limiting the number of nodes configured=%d (max=%d)" %(len(node_list), LimitNodes)) while len(node_list) > LimitNodes: node_list.pop(len(node_list)-1) if StonithParams[0] == "hostlist": StonithParams[1] = string.join(node_list, " ") # alt_list = [] # for node in node_list: # alt_list.append(string.lower(node)) # node_list = alt_list Environment = CtsLab(node_list) Environment["HAdir"] = HAdir Environment["ClobberCIB"] = ClobberCIB Environment["CIBfilename"] = CIBfilename Environment["CIBResource"] = CIBResource Environment["LogFileName"] = LogFile Environment["DoStonith"] = DoStonith Environment["SyslogFacility"] = LogFacility Environment["DoStandby"] = DoStandby Environment["DoFencing"] = DoFencing Environment["ResCanStop"] = ResCanStop Environment["SuppressMonitoring"] = SuppressMonitoring Environment["XmitLoss"] = XmitLoss Environment["RecvLoss"] = RecvLoss Environment["IPBase"] = IPBase Environment["SuppressCib"] = SuppressCib Environment["DoBSC"] = 0 Environment["use_logd"] = 0 Environment["logfacility"] = LogFacility Environment["oprofile"] = oprofile Environment["warn-inactive"] = warn_inactive if config.Parameters.has_key("use_logd"): Environment["use_logd"] = 1 if Version == 2: from CM_LinuxHAv2 import LinuxHAv2 Environment['CMclass']=LinuxHAv2 if Version == 3: from CM_ais import crm_ais Environment['CMclass'] = crm_ais Environment["DoStonith"] = 0 Environment["DoFencing"] = 0 Environment["use_logd"] = 0 if HaveSeed: Environment["RandSeed"] = Seed Environment["reset"] = Stonith(sttype=StonithType, pName=StonithParams[0], pValue=StonithParams[1]) if DoBSC: Environment["DoBSC"] = 1 Environment["ClobberCIB"] = 1 Environment["CIBResource"] = 0 Environment["logger"] = (FileLog(Environment), StdErrLog(Environment)) scenario = Scenario([ BasicSanityCheck(Environment) ]) else: scenario = Scenario( [ InitClusterManager(Environment), PacketLoss(Environment)]) Environment.SupplyDefaults() # Your basic start up the world type of test scenario... #scenario = Scenario( #[ InitClusterManager(Environment) #, PingFest(Environment)]) # Create the Cluster Manager object cm = Environment['CMclass'](Environment) if TruncateLog: cm.log("Truncating %s" % LogFile) lf = open(LogFile, "w"); if lf != None: lf.truncate(0) lf.close() cm.log(">>>>>>>>>>>>>>>> BEGINNING " + repr(NumIter) + " TESTS ") cm.log("Version: %d" % Version) cm.log("HA configuration directory: " + Environment["HAdir"]) cm.log("System log files: " + Environment["LogFileName"]) cm.log("Enable Stonith: %d" % Environment["DoStonith"]) cm.log("Enable Fencing: %d" % Environment["DoFencing"]) cm.log("Enable Standby: %d" % Environment["DoStandby"]) cm.log("Enable Resources: %d" % Environment["CIBResource"]) if Environment.has_key("SuppressMonitoring") \ and Environment["SuppressMonitoring"]: cm.log("Resource Monitoring is disabled") cm.ns.WaitForAllNodesToComeUp(config.Parameters["node"]) cm.log("Cluster nodes: ") for node in config.Parameters["node"]: (rc, lines) = cm.rsh.remote_py(node, "os", "system", "@sbindir@/crm_uuid") if not lines: cm.log(" * %s: __undefined_uuid__" % node) else: out=lines[0] out = out[:-1] cm.log(" * %s: %s" % (node, out)) Audits = AuditList(cm) Tests = [] if Environment["DoBSC"]: test = BSC_AddResource(cm) Tests.append(test) elif TestCase != None: for test in TestList(cm): if test.name == TestCase: Tests.append(test) if Tests == []: usage("--choose: No applicable/valid tests chosen") else: Tests = TestList(cm) if ListTests == 1 : cm.log("Total %d tests"%len(Tests)) for test in Tests : cm.log(str(test.name)); sys.exit(0) tests = RandomTests(scenario, cm, Tests, Audits) Environment.RandomTests = tests try : overall, detailed = tests.run(NumIter) except : cm.Env.log("Exception by %s" % sys.exc_info()[0]) for logmethod in Environment["logger"]: traceback.print_exc(50, logmethod) tests.summarize() if tests.Stats["failure"] > 0: sys.exit(tests.Stats["failure"]) elif tests.Stats["success"] != NumIter: cm.Env.log("No failure count but success != requested iterations") sys.exit(1) diff --git a/include/crm/ais.h b/include/crm/ais.h index d14bd5206d..60be30ea76 100644 --- a/include/crm/ais.h +++ b/include/crm/ais.h @@ -1,39 +1,34 @@ /* * 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 CRM_AIS__H #define CRM_AIS__H #include #include #include -#if SUPPORT_AIS -#include -#include -#include -#endif #include extern enum crm_ais_msg_types text2msg_type(const char *text); extern enum crm_ais_msg_types crm_system_type; extern char *get_ais_data(AIS_Message *msg); extern gboolean check_message_sanity(AIS_Message *msg, char *data); #endif diff --git a/include/crm/ais_common.h b/include/crm/ais_common.h index abad7f310b..483fbcaa62 100644 --- a/include/crm/ais_common.h +++ b/include/crm/ais_common.h @@ -1,304 +1,339 @@ /* * 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 CRM_AIS_COMMON__H #define CRM_AIS_COMMON__H #include #include #include #if SUPPORT_AIS -# include -# include -# include +# ifdef AIS_WHITETANK +/* cheap hacks for building against the stable series of openais */ + +# include + +enum service_types { + EVS_SERVICE = 0, + CLM_SERVICE = 1, + AMF_SERVICE = 2, + CKPT_SERVICE = 3, + EVT_SERVICE = 4, + LCK_SERVICE = 5, + MSG_SERVICE = 6, + CFG_SERVICE = 7, + CPG_SERVICE = 8 +}; + +typedef struct { + int size; __attribute__((aligned(8))) + int id __attribute__((aligned(8))); + SaAisErrorT error __attribute__((aligned(8))); +} mar_res_header_t __attribute__((aligned(8))); + +typedef struct { + int size __attribute__((aligned(8))); + int id __attribute__((aligned(8))); +} mar_req_header_t __attribute__((aligned(8))); + +extern SaAisErrorT saSendReceiveReply ( + int s, void *requestMessage, int requestLen, void *responseMessage, int responseLen); +extern SaAisErrorT saRecvRetry (int s, void *msg, size_t len); +extern SaAisErrorT saServiceConnect (int *responseOut, int *callbackOut, enum service_types service); + +# else +# include +# include +# include +# endif #else typedef struct { int size __attribute__((aligned(8))); int id __attribute__((aligned(8))); } mar_req_header_t __attribute__((aligned(8))); typedef struct { int size; __attribute__((aligned(8))) int id __attribute__((aligned(8))); int error __attribute__((aligned(8))); } mar_res_header_t __attribute__((aligned(8))); #endif #define MAX_NAME 256 #define AIS_IPC_NAME "ais-crm-ipc" #define CRM_NODE_LOST "lost" #define CRM_NODE_MEMBER "member" #define CRM_NODE_ACTIVE CRM_NODE_MEMBER #define CRM_NODE_INACTIVE CRM_NODE_LOST #define CRM_NODE_EVICTED "evicted" typedef struct crm_ais_host_s AIS_Host; typedef struct crm_ais_msg_s AIS_Message; enum crm_ais_msg_class { crm_class_cluster = 0, crm_class_members = 1, crm_class_notify = 2, }; /* order here matters - its used to index into the crm_children array */ enum crm_ais_msg_types { - crm_msg_none = 0, - crm_msg_ais = 1, - crm_msg_lrmd = 2, - crm_msg_cib = 3, - crm_msg_crmd = 4, - crm_msg_attrd = 5, + crm_msg_none = 0, + crm_msg_ais = 1, + crm_msg_lrmd = 2, + crm_msg_cib = 3, + crm_msg_crmd = 4, + crm_msg_attrd = 5, crm_msg_stonithd = 6, - crm_msg_te = 7, - crm_msg_pe = 8, + crm_msg_te = 7, + crm_msg_pe = 8, }; enum crm_proc_flag { crm_proc_none = 0x00000001, crm_proc_ais = 0x00000002, crm_proc_lrmd = 0x00000010, crm_proc_cib = 0x00000100, crm_proc_crmd = 0x00000200, crm_proc_attrd = 0x00001000, crm_proc_stonithd = 0x00002000, crm_proc_pe = 0x00010000, crm_proc_te = 0x00020000, }; typedef struct crm_peer_node_s { unsigned int id; unsigned long long born; unsigned long long last_seen; int32_t votes; uint32_t processes; char *uname; char *state; char *uuid; char *addr; char *version; } crm_node_t; struct crm_ais_host_s { uint32_t id; uint32_t pid; gboolean local; enum crm_ais_msg_types type; uint32_t size; char uname[256]; } __attribute__((packed)); struct crm_ais_msg_s { mar_res_header_t header __attribute__((aligned(8))); uint32_t id; gboolean is_compressed; AIS_Host host; AIS_Host sender; uint32_t size; uint32_t compressed_size; /* 584 bytes */ char data[0]; } __attribute__((packed)); #if SUPPORT_AIS static inline const char *ais_error2text(SaAisErrorT error) { const char *text = "unknown"; switch(error) { case SA_AIS_OK: text = "None"; break; case SA_AIS_ERR_LIBRARY: text = "Library error"; break; case SA_AIS_ERR_VERSION: text = "Version error"; break; case SA_AIS_ERR_INIT: text = "Initialization error"; break; case SA_AIS_ERR_TIMEOUT: text = "Timeout"; break; case SA_AIS_ERR_TRY_AGAIN: text = "Try again"; break; case SA_AIS_ERR_INVALID_PARAM: text = "Invalid parameter"; break; case SA_AIS_ERR_NO_MEMORY: text = "No memory"; break; case SA_AIS_ERR_BAD_HANDLE: text = "Bad handle"; break; case SA_AIS_ERR_BUSY: text = "Busy"; break; case SA_AIS_ERR_ACCESS: text = "Access error"; break; case SA_AIS_ERR_NOT_EXIST: text = "Doesn't exist"; break; case SA_AIS_ERR_NAME_TOO_LONG: text = "Name too long"; break; case SA_AIS_ERR_EXIST: text = "Exists"; break; case SA_AIS_ERR_NO_SPACE: text = "No space"; break; case SA_AIS_ERR_INTERRUPT: text = "Interrupt"; break; case SA_AIS_ERR_NAME_NOT_FOUND: text = "Name not found"; break; case SA_AIS_ERR_NO_RESOURCES: text = "No resources"; break; case SA_AIS_ERR_NOT_SUPPORTED: text = "Not supported"; break; case SA_AIS_ERR_BAD_OPERATION: text = "Bad operation"; break; case SA_AIS_ERR_FAILED_OPERATION: text = "Failed operation"; break; case SA_AIS_ERR_MESSAGE_ERROR: text = "Message error"; break; case SA_AIS_ERR_QUEUE_FULL: text = "Queue full"; break; case SA_AIS_ERR_QUEUE_NOT_AVAILABLE: text = "Queue not available"; break; case SA_AIS_ERR_BAD_FLAGS: text = "Bad flags"; break; case SA_AIS_ERR_TOO_BIG: text = "To big"; break; case SA_AIS_ERR_NO_SECTIONS: text = "No sections"; break; } return text; } #endif static inline const char *msg_type2text(enum crm_ais_msg_types type) { const char *text = "unknown"; switch(type) { case crm_msg_none: text = "unknown"; break; case crm_msg_ais: text = "ais"; break; case crm_msg_cib: text = "cib"; break; case crm_msg_crmd: text = "crmd"; break; case crm_msg_pe: text = "pengine"; break; case crm_msg_te: text = "tengine"; break; case crm_msg_lrmd: text = "lrmd"; break; case crm_msg_attrd: text = "attrd"; break; case crm_msg_stonithd: text = "stonithd"; break; } return text; } static inline const char *peer2text(enum crm_proc_flag proc) { const char *text = "unknown"; switch(proc) { case crm_proc_none: text = "unknown"; break; case crm_proc_ais: text = "ais"; break; case crm_proc_cib: text = "cib"; break; case crm_proc_crmd: text = "crmd"; break; case crm_proc_pe: text = "pengine"; break; case crm_proc_te: text = "tengine"; break; case crm_proc_lrmd: text = "lrmd"; break; case crm_proc_attrd: text = "attrd"; break; case crm_proc_stonithd: text = "stonithd"; break; } return text; } static inline const char *ais_dest(struct crm_ais_host_s *host) { if(host->local) { return "local"; } else if(host->size > 0) { return host->uname; } else { return ""; } } #define ais_data_len(msg) (msg->is_compressed?msg->compressed_size:msg->size) #endif diff --git a/lib/crm/common/ais.c b/lib/crm/common/ais.c index 387a57271a..7f8fc36f4d 100644 --- a/lib/crm/common/ais.c +++ b/lib/crm/common/ais.c @@ -1,496 +1,505 @@ /* * 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 "stack.h" #include enum crm_ais_msg_types text2msg_type(const char *text) { int type = crm_msg_none; CRM_CHECK(text != NULL, return type); if(safe_str_eq(text, "ais")) { type = crm_msg_ais; } else if(safe_str_eq(text, "crm_plugin")) { type = crm_msg_ais; } else if(safe_str_eq(text, CRM_SYSTEM_CIB)) { type = crm_msg_cib; } else if(safe_str_eq(text, CRM_SYSTEM_CRMD)) { type = crm_msg_crmd; } else if(safe_str_eq(text, CRM_SYSTEM_DC)) { type = crm_msg_crmd; } else if(safe_str_eq(text, CRM_SYSTEM_TENGINE)) { type = crm_msg_te; } else if(safe_str_eq(text, CRM_SYSTEM_PENGINE)) { type = crm_msg_pe; } else if(safe_str_eq(text, CRM_SYSTEM_LRMD)) { type = crm_msg_lrmd; } else if(safe_str_eq(text, CRM_SYSTEM_STONITHD)) { type = crm_msg_stonithd; + } else if(safe_str_eq(text, "attrd")) { + type = crm_msg_attrd; } else { crm_debug_2("Unknown message type: %s", text); } return type; } char *get_ais_data(AIS_Message *msg) { int rc = BZ_OK; char *uncompressed = NULL; unsigned int new_size = msg->size; if(msg->is_compressed == FALSE) { crm_debug_2("Returning uncompressed message data"); uncompressed = strdup(msg->data); } else { crm_debug_2("Decompressing message data"); crm_malloc0(uncompressed, new_size); rc = BZ2_bzBuffToBuffDecompress( uncompressed, &new_size, msg->data, msg->compressed_size, 1, 0); CRM_ASSERT(rc = BZ_OK); CRM_ASSERT(new_size == msg->size); } return uncompressed; } #if SUPPORT_AIS int ais_fd_sync = -1; static int ais_fd_async = -1; /* never send messages via this channel */ GFDSource *ais_source = NULL; GFDSource *ais_source_sync = NULL; -struct res_overlay { - mar_res_header_t header __attribute((aligned(8))); -/* char buf[4096]; */ -}; - gboolean send_ais_text(int class, const char *data, gboolean local, const char *node, enum crm_ais_msg_types dest) { int retries = 0; static int msg_id = 0; static int local_pid = 0; int rc = SA_AIS_OK; mar_res_header_t header; AIS_Message *ais_msg = NULL; enum crm_ais_msg_types sender = text2msg_type(crm_system_name); if(local_pid == 0) { local_pid = getpid(); } CRM_CHECK(data != NULL, return FALSE); crm_malloc0(ais_msg, sizeof(AIS_Message)); ais_msg->id = msg_id++; ais_msg->header.id = class; ais_msg->host.type = dest; ais_msg->host.local = local; if(node) { ais_msg->host.size = strlen(node); memset(ais_msg->host.uname, 0, MAX_NAME); memcpy(ais_msg->host.uname, node, ais_msg->host.size); ais_msg->host.id = 0; } else { ais_msg->host.size = 0; memset(ais_msg->host.uname, 0, MAX_NAME); ais_msg->host.id = 0; } ais_msg->sender.type = sender; ais_msg->sender.pid = local_pid; ais_msg->sender.size = 0; memset(ais_msg->sender.uname, 0, MAX_NAME); ais_msg->sender.id = 0; ais_msg->size = 1 + strlen(data); if(ais_msg->size < 5120) { failback: crm_realloc(ais_msg, sizeof(AIS_Message) + ais_msg->size); memcpy(ais_msg->data, data, ais_msg->size); } else { char *compressed = NULL; char *uncompressed = crm_strdup(data); unsigned int len = (ais_msg->size * 1.1) + 600; /* recomended size */ crm_debug_5("Compressing message payload"); crm_malloc0(compressed, len); rc = BZ2_bzBuffToBuffCompress( compressed, &len, uncompressed, ais_msg->size, 3, 0, 30); crm_free(uncompressed); if(rc != BZ_OK) { crm_err("Compression failed: %d", rc); crm_free(compressed); goto failback; } crm_realloc(ais_msg, sizeof(AIS_Message) + len + 1); memcpy(ais_msg->data, compressed, len); crm_free(compressed); ais_msg->is_compressed = TRUE; ais_msg->compressed_size = len; crm_debug("Compression details: %d -> %d", ais_msg->size, ais_data_len(ais_msg)); } ais_msg->header.size = sizeof(AIS_Message) + ais_data_len(ais_msg); crm_debug("Sending%s message %d to %s.%s (data=%d, total=%d)", ais_msg->is_compressed?" compressed":"", ais_msg->id, ais_dest(&(ais_msg->host)), msg_type2text(dest), ais_data_len(ais_msg), ais_msg->header.size); retry: errno = 0; rc = saSendReceiveReply(ais_fd_sync, ais_msg, ais_msg->header.size, &header, sizeof (mar_res_header_t)); if(rc == SA_AIS_OK) { - CRM_CHECK(header.error == 0, rc = header.error); + CRM_CHECK(header.size == sizeof (mar_res_header_t), + crm_err("Odd message: id=%d, size=%d, error=%d", + header.id, header.size, header.error)); + CRM_CHECK(header.error == SA_AIS_OK, rc = header.error); } if(rc == SA_AIS_ERR_TRY_AGAIN && retries < 20) { retries++; crm_info("Peer overloaded: Re-sending message (Attempt %d of 20)", retries); mssleep(retries * 100); /* Proportional back off */ goto retry; } if(rc != SA_AIS_OK) { cl_perror("Sending message %d: FAILED (rc=%d): %s", ais_msg->id, rc, ais_error2text(rc)); ais_fd_async = -1; } else { crm_debug_4("Message %d: sent", ais_msg->id); } crm_free(ais_msg); return (rc == SA_AIS_OK); } gboolean send_ais_message(crm_data_t *msg, gboolean local, const char *node, enum crm_ais_msg_types dest) { gboolean rc = TRUE; char *data = NULL; if(ais_fd_async < 0 || ais_source == NULL) { crm_err("Not connected to AIS"); return FALSE; } if(cl_get_string(msg, F_XML_TAGNAME) == NULL) { ha_msg_add(msg, F_XML_TAGNAME, "ais_msg"); } data = dump_xml_unformatted(msg); rc = send_ais_text(0, data, local, node, dest); crm_free(data); return rc; } void terminate_ais_connection(void) { close(ais_fd_sync); close(ais_fd_async); crm_notice("Disconnected from AIS"); /* G_main_del_fd(ais_source); */ /* G_main_del_fd(ais_source_sync); */ } static gboolean ais_dispatch(int sender, gpointer user_data) { char *data = NULL; char *uncompressed = NULL; AIS_Message *msg = NULL; SaAisErrorT rc = SA_AIS_OK; mar_res_header_t *header = NULL; static int header_len = sizeof(mar_res_header_t); gboolean (*dispatch)(AIS_Message*,char*,int) = user_data; crm_malloc0(header, header_len); errno = 0; rc = saRecvRetry(sender, header, header_len); if (rc != SA_AIS_OK) { cl_perror("Receiving message header failed: (%d) %s", rc, ais_error2text(rc)); goto bail; } else if(header->size == header_len) { - crm_err("Empty message: error=%d", header->error); + crm_err("Empty message: id=%d, size=%d, error=%d, header_len=%d", + header->id, header->size, header->error, header_len); goto done; } else if(header->size == 0 || header->size < header_len) { crm_err("Mangled header: size=%d, header=%d, error=%d", header->size, header_len, header->error); goto done; } else if(header->error != 0) { crm_err("Header contined error: %d", header->error); } crm_debug_2("Looking for %d (%d - %d) more bytes", header->size - header_len, header->size, header_len); crm_realloc(header, header->size); /* Use a char* so we can store the remainder into an offset */ data = (char*)header; errno = 0; rc = saRecvRetry(sender, data+header_len, header->size - header_len); msg = (AIS_Message*)data; if (rc != SA_AIS_OK) { cl_perror("Receiving message body failed: (%d) %s", rc, ais_error2text(rc)); goto bail; } crm_debug_3("Got new%s message (size=%d, %d, %d)", msg->is_compressed?" compressed":"", ais_data_len(msg), msg->size, msg->compressed_size); data = msg->data; if(msg->is_compressed) { int rc = BZ_OK; unsigned int new_size = msg->size; if(check_message_sanity(msg, NULL) == FALSE) { goto badmsg; } crm_debug_5("Decompressing message data"); crm_malloc0(uncompressed, new_size); rc = BZ2_bzBuffToBuffDecompress( uncompressed, &new_size, data, msg->compressed_size, 1, 0); if(rc != BZ_OK) { crm_err("Decompression failed: %d", rc); crm_free(uncompressed); goto badmsg; } CRM_ASSERT(rc == BZ_OK); CRM_ASSERT(new_size == msg->size); data = uncompressed; } else if(check_message_sanity(msg, data) == FALSE) { goto badmsg; } else if(safe_str_eq("identify", data)) { int pid = getpid(); char *pid_s = crm_itoa(pid); send_ais_text(0, pid_s, TRUE, NULL, crm_msg_ais); crm_free(pid_s); goto done; } if(msg->header.id == crm_class_members) { crm_data_t *xml = string2xml(data); if(xml != NULL) { const char *seq_s = crm_element_value(xml, "id"); unsigned long seq = crm_int_helper(seq_s, NULL); crm_info("Processing membership %ld/%s", seq, seq_s); /* crm_log_xml_debug(xml, __PRETTY_FUNCTION__); */ xml_child_iter(xml, node, crm_update_ais_node(node, seq)); crm_calculate_quorum(); } else { crm_warn("Invalid peer update: %s", data); } free_xml(xml); } if(dispatch != NULL) { dispatch(msg, data, sender); } done: crm_free(uncompressed); crm_free(msg); return TRUE; badmsg: crm_err("Invalid message (id=%d, dest=%s:%s, from=%s:%s.%d):" " min=%d, total=%d, size=%d, bz2_size=%d", msg->id, ais_dest(&(msg->host)), msg_type2text(msg->host.type), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, (int)sizeof(AIS_Message), msg->header.size, msg->size, msg->compressed_size); goto done; bail: crm_err("AIS connection failed"); return FALSE; } static void ais_destroy(gpointer user_data) { crm_err("AIS connection terminated"); ais_fd_sync = -1; exit(1); } gboolean init_ais_connection( gboolean (*dispatch)(AIS_Message*,char*,int), void (*destroy)(gpointer), char **our_uuid, char **our_uname) { int rc = SA_AIS_OK; struct utsname name; if(our_uname != NULL) { if(uname(&name) < 0) { cl_perror("uname(2) call failed"); exit(100); } *our_uname = crm_strdup(name.nodename); crm_notice("Local node name: %s", *our_uname); } if(our_uuid != NULL) { *our_uuid = crm_strdup(name.nodename); } /* 16 := CRM_SERVICE */ crm_info("Creating connection to our AIS plugin"); - rc = saServiceConnect (&ais_fd_async, &ais_fd_sync, 16); + rc = saServiceConnect (&ais_fd_sync, &ais_fd_async, 16); if (rc != SA_AIS_OK) { - crm_info("Connection to our AIS plugin failed"); + crm_info("Connection to our AIS plugin failed: %s (%d)", ais_error2text(rc), rc); return FALSE; } if(destroy == NULL) { crm_debug("Using the default destroy handler"); destroy = ais_destroy; } crm_info("AIS connection established"); #if 0 ais_source_sync = G_main_add_fd( G_PRIORITY_HIGH, ais_fd_sync, FALSE, ais_dispatch, dispatch, destroy); #endif - +#if AIS_WHITETANK + { + int pid = getpid(); + char *pid_s = crm_itoa(pid); + send_ais_text(0, pid_s, TRUE, NULL, crm_msg_ais); + crm_free(pid_s); + } +#endif + ais_source = G_main_add_fd( G_PRIORITY_HIGH, ais_fd_async, FALSE, ais_dispatch, dispatch, destroy); return TRUE; } gboolean check_message_sanity(AIS_Message *msg, char *data) { gboolean sane = TRUE; gboolean repaired = FALSE; int dest = msg->host.type; int tmp_size = msg->header.size - sizeof(AIS_Message); if(sane && msg->header.size == 0) { crm_warn("Message with no size"); sane = FALSE; } if(sane && msg->header.error != 0) { crm_warn("Message header contains an error: %d", msg->header.error); sane = FALSE; } if(sane && ais_data_len(msg) != tmp_size) { int cur_size = ais_data_len(msg); repaired = TRUE; if(msg->is_compressed) { msg->compressed_size = tmp_size; } else { msg->size = tmp_size; } crm_warn("Repaired message payload size %d -> %d", cur_size, tmp_size); } if(sane && ais_data_len(msg) == 0) { crm_warn("Message with no payload"); sane = FALSE; } if(sane && data && msg->is_compressed == FALSE) { int str_size = strlen(data) + 1; if(ais_data_len(msg) != str_size) { int lpc = 0; crm_warn("Message payload is corrupted: expected %d bytes, got %d", ais_data_len(msg), str_size); sane = FALSE; for(lpc = (str_size - 10); lpc < msg->size; lpc++) { if(lpc < 0) { lpc = 0; } crm_debug("bad_data[%d]: %d / '%c'", lpc, data[lpc], data[lpc]); } } } if(sane == FALSE) { crm_err("Invalid message %d: (dest=%s:%s, from=%s:%s.%d, compressed=%d, size=%d, total=%d)", msg->id, ais_dest(&(msg->host)), msg_type2text(dest), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->is_compressed, ais_data_len(msg), msg->header.size); } else if(repaired) { crm_err("Repaired message %d: (dest=%s:%s, from=%s:%s.%d, compressed=%d, size=%d, total=%d)", msg->id, ais_dest(&(msg->host)), msg_type2text(dest), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->is_compressed, ais_data_len(msg), msg->header.size); } else { crm_debug_3("Verfied message %d: (dest=%s:%s, from=%s:%s.%d, compressed=%d, size=%d, total=%d)", msg->id, ais_dest(&(msg->host)), msg_type2text(dest), ais_dest(&(msg->sender)), msg_type2text(msg->sender.type), msg->sender.pid, msg->is_compressed, ais_data_len(msg), msg->header.size); } return sane; } #endif diff --git a/lib/plugins/lrm/Makefile.am b/lib/plugins/lrm/Makefile.am index 45c9765926..cfc8ae9ed3 100644 --- a/lib/plugins/lrm/Makefile.am +++ b/lib/plugins/lrm/Makefile.am @@ -1,43 +1,42 @@ # # Author: Sun Jiang Dong # Copyright (c) 2004 International Business Machines # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # MAINTAINERCLEANFILES = Makefile.in LRM_DIR = lrm INCLUDES = -I$(top_builddir)/include -I$(top_srcdir)/include \ -I$(top_builddir)/linux-ha -I$(top_srcdir)/linux-ha \ -I$(top_builddir)/libltdl -I$(top_srcdir)/libltdl halibdir = $(libdir)/@HB_PKG@ havarlibdir = $(localstatedir)/lib/@HB_PKG@ COMMONLIBS = -lplumb \ $(GLIBLIB) lrmdir = $(HA_VARLIBDIR)/$(HB_PKG)/$(LRM_DIR) plugindir = $(halibdir)/plugins/RAExec -apigid = @HA_APIGID@ plugin_LTLIBRARIES = stonith.la stonith_la_SOURCES = raexecstonith.c stonith_la_LDFLAGS = -lpils -export-dynamic -module -avoid-version \ -L$(top_builddir)/lib/fencing -lstonithd -lstonith -llrm install-exec-local: $(mkinstalldirs) $(DESTDIR)$(lrmdir) - -chgrp $(apigid) $(DESTDIR)/$(lrmdir) + -chgrp $(HA_APIGROUP) $(DESTDIR)/$(lrmdir) chmod 770 $(DESTDIR)/$(lrmdir)