diff --git a/crm/ais/utils.h b/crm/ais/utils.h index cedb48236d..99ff898630 100644 --- a/crm/ais/utils.h +++ b/crm/ais/utils.h @@ -1,192 +1,190 @@ /* * 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 "plugin.h" */ #define SIZEOF(a) (sizeof(a) / sizeof(a[0])) -#define CRM_MESSAGE_IPC_ACK 1 -#define CRM_SERVICE 9 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/include/crm/ais_common.h b/include/crm/ais_common.h index 483fbcaa62..be33bdea09 100644 --- a/include/crm/ais_common.h +++ b/include/crm/ais_common.h @@ -1,339 +1,341 @@ /* * 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 # 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" +#define CRM_MESSAGE_IPC_ACK 1 +#define CRM_SERVICE 9 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_stonithd = 6, 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 403a716b79..0521c15129 100644 --- a/lib/crm/common/ais.c +++ b/lib/crm/common/ais.c @@ -1,503 +1,503 @@ /* * 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; 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.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: 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 && msg->size > 0) { 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_sync, &ais_fd_async, 16); + rc = saServiceConnect (&ais_fd_sync, &ais_fd_async, CRM_SERVICE); if (rc != SA_AIS_OK) { 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 { int pid = getpid(); char *pid_s = crm_itoa(pid); send_ais_text(0, pid_s, TRUE, NULL, crm_msg_ais); crm_free(pid_s); } 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