diff --git a/lib/cluster/corosync.c b/lib/cluster/corosync.c
index 86f7061e97..10ac46ddef 100644
--- a/lib/cluster/corosync.c
+++ b/lib/cluster/corosync.c
@@ -1,560 +1,555 @@
/*
* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser 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 library 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
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <bzlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <crm/common/ipc.h>
#include <crm/cluster/internal.h>
#include <crm/common/mainloop.h>
#include <sys/utsname.h>
#include <qb/qbipcc.h>
#include <qb/qbutil.h>
#include <corosync/corodefs.h>
#include <corosync/corotypes.h>
#include <corosync/hdb.h>
#include <corosync/cfg.h>
#include <corosync/cmap.h>
#include <corosync/quorum.h>
#include <crm/msg_xml.h>
quorum_handle_t pcmk_quorum_handle = 0;
gboolean(*quorum_app_callback) (unsigned long long seq, gboolean quorate) = NULL;
/*
* CFG functionality stolen from node_name() in corosync-quorumtool.c
* This resolves the first address assigned to a node and returns the name or IP address.
*/
char *
corosync_node_name(uint64_t /*cmap_handle_t */ cmap_handle, uint32_t nodeid)
{
int lpc = 0;
int rc = CS_OK;
int retries = 0;
char *name = NULL;
cmap_handle_t local_handle = 0;
/* nodeid == 0 == CMAN_NODEID_US */
if (nodeid == 0) {
nodeid = get_local_nodeid(0);
}
if (cmap_handle == 0 && local_handle == 0) {
retries = 0;
crm_trace("Initializing CMAP connection");
do {
rc = cmap_initialize(&local_handle);
if (rc != CS_OK) {
retries++;
crm_debug("API connection setup failed: %s. Retrying in %ds", cs_strerror(rc),
retries);
sleep(retries);
}
} while (retries < 5 && rc != CS_OK);
if (rc != CS_OK) {
crm_warn("Could not connect to Cluster Configuration Database API, error %s",
cs_strerror(rc));
local_handle = 0;
}
}
if (cmap_handle == 0) {
cmap_handle = local_handle;
}
while (name == NULL && cmap_handle != 0) {
uint32_t id = 0;
char *key = NULL;
key = g_strdup_printf("nodelist.node.%d.nodeid", lpc);
rc = cmap_get_uint32(cmap_handle, key, &id);
crm_trace("Checking %u vs %u from %s", nodeid, id, key);
g_free(key);
if (rc != CS_OK) {
break;
}
if (nodeid == id) {
crm_trace("Searching for node name for %u in nodelist.node.%d %s", nodeid, lpc, name);
if (name == NULL) {
key = g_strdup_printf("nodelist.node.%d.ring0_addr", lpc);
rc = cmap_get_string(cmap_handle, key, &name);
crm_trace("%s = %s", key, name);
if (node_name_is_valid(key, name) == FALSE) {
free(name);
name = NULL;
}
g_free(key);
}
if (name == NULL) {
key = g_strdup_printf("nodelist.node.%d.name", lpc);
rc = cmap_get_string(cmap_handle, key, &name);
crm_trace("%s = %s %d", key, name, rc);
g_free(key);
}
break;
}
lpc++;
}
if(local_handle) {
cmap_finalize(local_handle);
}
if (name == NULL) {
crm_notice("Unable to get node name for nodeid %u", nodeid);
}
return name;
}
void
terminate_cs_connection(crm_cluster_t *cluster)
{
crm_notice("Disconnecting from Corosync");
cluster_disconnect_cpg(cluster);
if (pcmk_quorum_handle) {
crm_trace("Disconnecting quorum");
quorum_finalize(pcmk_quorum_handle);
pcmk_quorum_handle = 0;
} else {
crm_info("No Quorum connection");
}
}
int ais_membership_timer = 0;
gboolean ais_membership_force = FALSE;
static int
pcmk_quorum_dispatch(gpointer user_data)
{
int rc = 0;
rc = quorum_dispatch(pcmk_quorum_handle, CS_DISPATCH_ALL);
if (rc < 0) {
crm_err("Connection to the Quorum API failed: %d", rc);
pcmk_quorum_handle = 0;
return -1;
}
return 0;
}
static void
pcmk_quorum_notification(quorum_handle_t handle,
uint32_t quorate,
uint64_t ring_id, uint32_t view_list_entries, uint32_t * view_list)
{
int i;
GHashTableIter iter;
crm_node_t *node = NULL;
static gboolean init_phase = TRUE;
if (quorate != crm_have_quorum) {
crm_notice("Membership " U64T ": quorum %s (%lu)", ring_id,
quorate ? "acquired" : "lost", (long unsigned int)view_list_entries);
crm_have_quorum = quorate;
} else {
crm_info("Membership " U64T ": quorum %s (%lu)", ring_id,
quorate ? "retained" : "still lost", (long unsigned int)view_list_entries);
}
if (view_list_entries == 0 && init_phase) {
crm_info("Corosync membership is still forming, ignoring");
return;
}
init_phase = FALSE;
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &node)) {
node->last_seen = 0;
}
for (i = 0; i < view_list_entries; i++) {
uint32_t id = view_list[i];
char *name = NULL;
crm_debug("Member[%d] %u ", i, id);
node = crm_get_peer(id, NULL);
if (node->uname == NULL) {
crm_info("Obtaining name for new node %u", id);
name = corosync_node_name(0, id);
node = crm_get_peer(id, name);
}
crm_update_peer_state(__FUNCTION__, node, CRM_NODE_MEMBER, ring_id);
free(name);
}
crm_trace("Reaping unseen nodes...");
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &node)) {
if (node->last_seen != ring_id && node->state) {
crm_update_peer_state(__FUNCTION__, node, CRM_NODE_LOST, 0);
} else if (node->last_seen != ring_id) {
crm_info("State of node %s[%u] is still unknown", node->uname, node->id);
}
}
if (quorum_app_callback) {
quorum_app_callback(ring_id, quorate);
}
}
quorum_callbacks_t quorum_callbacks = {
.quorum_notify_fn = pcmk_quorum_notification,
};
gboolean
init_quorum_connection(gboolean(*dispatch) (unsigned long long, gboolean),
void (*destroy) (gpointer))
{
int rc = -1;
int fd = 0;
int quorate = 0;
uint32_t quorum_type = 0;
struct mainloop_fd_callbacks quorum_fd_callbacks;
quorum_fd_callbacks.dispatch = pcmk_quorum_dispatch;
quorum_fd_callbacks.destroy = destroy;
crm_debug("Configuring Pacemaker to obtain quorum from Corosync");
rc = quorum_initialize(&pcmk_quorum_handle, &quorum_callbacks, &quorum_type);
if (rc != CS_OK) {
crm_err("Could not connect to the Quorum API: %d\n", rc);
goto bail;
} else if (quorum_type != QUORUM_SET) {
crm_err("Corosync quorum is not configured\n");
goto bail;
}
rc = quorum_getquorate(pcmk_quorum_handle, &quorate);
if (rc != CS_OK) {
crm_err("Could not obtain the current Quorum API state: %d\n", rc);
goto bail;
}
crm_notice("Quorum %s", quorate ? "acquired" : "lost");
quorum_app_callback = dispatch;
crm_have_quorum = quorate;
rc = quorum_trackstart(pcmk_quorum_handle, CS_TRACK_CHANGES | CS_TRACK_CURRENT);
if (rc != CS_OK) {
crm_err("Could not setup Quorum API notifications: %d\n", rc);
goto bail;
}
rc = quorum_fd_get(pcmk_quorum_handle, &fd);
if (rc != CS_OK) {
crm_err("Could not obtain the Quorum API connection: %d\n", rc);
goto bail;
}
mainloop_add_fd("quorum", G_PRIORITY_HIGH, fd, dispatch, &quorum_fd_callbacks);
corosync_initialize_nodelist(NULL, FALSE, NULL);
bail:
if (rc != CS_OK) {
quorum_finalize(pcmk_quorum_handle);
return FALSE;
}
return TRUE;
}
gboolean
init_cs_connection(crm_cluster_t * cluster)
{
int retries = 0;
while (retries < 5) {
int rc = init_cs_connection_once(cluster);
retries++;
switch (rc) {
case CS_OK:
return TRUE;
break;
case CS_ERR_TRY_AGAIN:
case CS_ERR_QUEUE_FULL:
sleep(retries);
break;
default:
return FALSE;
}
}
crm_err("Could not connect to corosync after %d retries", retries);
return FALSE;
}
gboolean
init_cs_connection_once(crm_cluster_t * cluster)
{
- const char *uuid = NULL;
crm_node_t *peer = NULL;
enum cluster_type_e stack = get_cluster_type();
crm_peer_init();
/* Here we just initialize comms */
if (stack != pcmk_cluster_corosync) {
crm_err("Invalid cluster type: %s (%d)", name_for_cluster_type(stack), stack);
return FALSE;
}
if (cluster_connect_cpg(cluster) == FALSE) {
return FALSE;
}
crm_info("Connection to '%s': established", name_for_cluster_type(stack));
cluster->nodeid = get_local_nodeid(0);
if(cluster->nodeid == 0) {
crm_err("Could not establish local nodeid");
return FALSE;
}
cluster->uname = get_node_name(0);
if(cluster->uname == NULL) {
crm_err("Could not establish local node name");
return FALSE;
}
/* Ensure the local node always exists */
peer = crm_get_peer(cluster->nodeid, cluster->uname);
- uuid = get_corosync_uuid(peer);
-
- if(uuid) {
- cluster->uuid = strdup(uuid);
- }
+ cluster->uuid = get_corosync_uuid(peer);
return TRUE;
}
gboolean
check_message_sanity(const AIS_Message * msg, const char *data)
{
gboolean sane = TRUE;
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 != CS_OK) {
crm_warn("Message header contains an error: %d", msg->header.error);
sane = FALSE;
}
if (sane && ais_data_len(msg) != tmp_size) {
crm_warn("Message payload size is incorrect: expected %d, got %d", ais_data_len(msg),
tmp_size);
sane = TRUE;
}
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.%u, 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_trace
("Verfied message %d: (dest=%s:%s, from=%s:%s.%u, 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;
}
enum cluster_type_e
find_corosync_variant(void)
{
int rc = CS_OK;
cmap_handle_t handle;
rc = cmap_initialize(&handle);
switch(rc) {
case CS_OK:
break;
case CS_ERR_SECURITY:
crm_debug("Failed to initialize the cmap API: Permission denied (%d)", rc);
/* It's there, we just can't talk to it.
* Good enough for us to identify as 'corosync'
*/
return pcmk_cluster_corosync;
default:
crm_info("Failed to initialize the cmap API: %s (%d)",
ais_error2text(rc), rc);
return pcmk_cluster_unknown;
}
cmap_finalize(handle);
return pcmk_cluster_corosync;
}
gboolean
crm_is_corosync_peer_active(const crm_node_t * node)
{
if (node == NULL) {
crm_trace("NULL");
return FALSE;
} else if (safe_str_neq(node->state, CRM_NODE_MEMBER)) {
crm_trace("%s: state=%s", node->uname, node->state);
return FALSE;
} else if ((node->processes & crm_proc_cpg) == 0) {
crm_trace("%s: processes=%.16x", node->uname, node->processes);
return FALSE;
}
return TRUE;
}
gboolean
corosync_initialize_nodelist(void *cluster, gboolean force_member, xmlNode * xml_parent)
{
int lpc = 0;
int rc = CS_OK;
int retries = 0;
gboolean any = FALSE;
cmap_handle_t cmap_handle;
do {
rc = cmap_initialize(&cmap_handle);
if (rc != CS_OK) {
retries++;
crm_debug("API connection setup failed: %s. Retrying in %ds", cs_strerror(rc),
retries);
sleep(retries);
}
} while (retries < 5 && rc != CS_OK);
if (rc != CS_OK) {
crm_warn("Could not connect to Cluster Configuration Database API, error %d", rc);
return FALSE;
}
crm_peer_init();
crm_trace("Initializing corosync nodelist");
for (lpc = 0;; lpc++) {
uint32_t nodeid = 0;
char *name = NULL;
char *key = NULL;
key = g_strdup_printf("nodelist.node.%d.nodeid", lpc);
rc = cmap_get_uint32(cmap_handle, key, &nodeid);
g_free(key);
if (rc != CS_OK) {
break;
}
name = corosync_node_name(cmap_handle, nodeid);
if (name != NULL) {
GHashTableIter iter;
crm_node_t *node = NULL;
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &node)) {
if(node && node->uname && strcasecmp(node->uname, name) == 0) {
if (node->id && node->id != nodeid) {
crm_crit("Nodes %u and %u share the same name '%s': shutting down", node->id,
nodeid, name);
crm_exit(DAEMON_RESPAWN_STOP);
}
}
}
}
if (nodeid > 0 || name != NULL) {
crm_trace("Initializing node[%d] %u = %s", lpc, nodeid, name);
crm_get_peer(nodeid, name);
}
if (nodeid > 0 && name != NULL) {
any = TRUE;
if (xml_parent) {
xmlNode *node = create_xml_node(xml_parent, XML_CIB_TAG_NODE);
crm_xml_add_int(node, XML_ATTR_ID, nodeid);
crm_xml_add(node, XML_ATTR_UNAME, name);
if (force_member) {
crm_xml_add(node, XML_ATTR_TYPE, CRM_NODE_MEMBER);
}
}
}
free(name);
}
cmap_finalize(cmap_handle);
return any;
}
diff --git a/lib/cluster/cpg.c b/lib/cluster/cpg.c
index 7862cab61c..e3a6057a34 100644
--- a/lib/cluster/cpg.c
+++ b/lib/cluster/cpg.c
@@ -1,675 +1,675 @@
/*
* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser 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 library 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
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <bzlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <crm/common/ipc.h>
#include <crm/cluster/internal.h>
#include <crm/common/mainloop.h>
#include <sys/utsname.h>
#include <qb/qbipcc.h>
#include <qb/qbutil.h>
#include <corosync/corodefs.h>
#include <corosync/corotypes.h>
#include <corosync/hdb.h>
#include <corosync/cpg.h>
#include <crm/msg_xml.h>
cpg_handle_t pcmk_cpg_handle = 0; /* TODO: Remove, use cluster.cpg_handle */
static bool cpg_evicted = FALSE;
gboolean(*pcmk_cpg_dispatch_fn) (int kind, const char *from, const char *data) = NULL;
#define cs_repeat(counter, max, code) do { \
code; \
if(rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) { \
counter++; \
crm_debug("Retrying operation after %ds", counter); \
sleep(counter); \
} else { \
break; \
} \
} while(counter < max)
void
cluster_disconnect_cpg(crm_cluster_t *cluster)
{
pcmk_cpg_handle = 0;
if (cluster->cpg_handle) {
crm_trace("Disconnecting CPG");
cpg_leave(cluster->cpg_handle, &cluster->group);
cpg_finalize(cluster->cpg_handle);
cluster->cpg_handle = 0;
} else {
crm_info("No CPG connection");
}
}
uint32_t get_local_nodeid(cpg_handle_t handle)
{
int rc = CS_OK;
int retries = 0;
static uint32_t local_nodeid = 0;
cpg_handle_t local_handle = handle;
cpg_callbacks_t cb = { };
if(local_nodeid != 0) {
return local_nodeid;
}
#if 0
/* Should not be necessary */
if(get_cluster_type() == pcmk_cluster_classic_ais) {
get_ais_details(&local_nodeid, NULL);
goto done;
}
#endif
if(handle == 0) {
crm_trace("Creating connection");
cs_repeat(retries, 5, rc = cpg_initialize(&local_handle, &cb));
}
if (rc == CS_OK) {
retries = 0;
crm_trace("Performing lookup");
cs_repeat(retries, 5, rc = cpg_local_get(local_handle, &local_nodeid));
}
if (rc != CS_OK) {
crm_err("Could not get local node id from the CPG API: %s (%d)", ais_error2text(rc), rc);
}
if(handle == 0) {
crm_trace("Closing connection");
cpg_finalize(local_handle);
}
crm_debug("Local nodeid is %u", local_nodeid);
return local_nodeid;
}
GListPtr cs_message_queue = NULL;
int cs_message_timer = 0;
static ssize_t crm_cs_flush(gpointer data);
static gboolean
crm_cs_flush_cb(gpointer data)
{
cs_message_timer = 0;
crm_cs_flush(data);
return FALSE;
}
#define CS_SEND_MAX 200
static ssize_t
crm_cs_flush(gpointer data)
{
int sent = 0;
ssize_t rc = 0;
int queue_len = 0;
static unsigned int last_sent = 0;
cpg_handle_t *handle = (cpg_handle_t *)data;
if (*handle == 0) {
crm_trace("Connection is dead");
return pcmk_ok;
}
queue_len = g_list_length(cs_message_queue);
if ((queue_len % 1000) == 0 && queue_len > 1) {
crm_err("CPG queue has grown to %d", queue_len);
} else if (queue_len == CS_SEND_MAX) {
crm_warn("CPG queue has grown to %d", queue_len);
}
if (cs_message_timer) {
/* There is already a timer, wait until it goes off */
crm_trace("Timer active %d", cs_message_timer);
return pcmk_ok;
}
while (cs_message_queue && sent < CS_SEND_MAX) {
struct iovec *iov = cs_message_queue->data;
errno = 0;
rc = cpg_mcast_joined(*handle, CPG_TYPE_AGREED, iov, 1);
if (rc != CS_OK) {
break;
}
sent++;
last_sent++;
crm_trace("CPG message sent, size=%d", iov->iov_len);
cs_message_queue = g_list_remove(cs_message_queue, iov);
free(iov[0].iov_base);
free(iov);
}
queue_len -= sent;
if (sent > 1 || cs_message_queue) {
crm_info("Sent %d CPG messages (%d remaining, last=%u): %s (%d)",
sent, queue_len, last_sent, ais_error2text(rc), rc);
} else {
crm_trace("Sent %d CPG messages (%d remaining, last=%u): %s (%d)",
sent, queue_len, last_sent, ais_error2text(rc), rc);
}
if (cs_message_queue) {
uint32_t delay_ms = 100;
if(rc != CS_OK) {
/* Proportionally more if sending failed but cap at 1s */
delay_ms = QB_MIN(1000, CS_SEND_MAX + (10 * queue_len));
}
cs_message_timer = g_timeout_add(delay_ms, crm_cs_flush_cb, data);
}
return rc;
}
gboolean
send_cpg_iov(struct iovec * iov)
{
static unsigned int queued = 0;
queued++;
crm_trace("Queueing CPG message %u (%d bytes)", queued, iov->iov_len);
cs_message_queue = g_list_append(cs_message_queue, iov);
crm_cs_flush(&pcmk_cpg_handle);
return TRUE;
}
static int
pcmk_cpg_dispatch(gpointer user_data)
{
int rc = 0;
crm_cluster_t *cluster = (crm_cluster_t*) user_data;
rc = cpg_dispatch(cluster->cpg_handle, CS_DISPATCH_ALL);
if (rc != CS_OK) {
crm_err("Connection to the CPG API failed: %s (%d)", ais_error2text(rc), rc);
cluster->cpg_handle = 0;
return -1;
} else if(cpg_evicted) {
crm_err("Evicted from CPG membership");
return -1;
}
return 0;
}
char *
pcmk_message_common_cs(cpg_handle_t handle, uint32_t nodeid, uint32_t pid, void *content,
uint32_t *kind, const char **from)
{
char *data = NULL;
AIS_Message *msg = (AIS_Message *) content;
if(handle) {
/* 'msg' came from CPG not the plugin
* Do filtering and field massaging
*/
uint32_t local_nodeid = get_local_nodeid(handle);
const char *local_name = get_local_node_name();
if (msg->sender.id > 0 && msg->sender.id != nodeid) {
crm_err("Nodeid mismatch from %d.%d: claimed nodeid=%u", nodeid, pid, msg->sender.id);
return NULL;
} else if (msg->host.id != 0 && (local_nodeid != msg->host.id)) {
/* Not for us */
crm_trace("Not for us: %u != %u", msg->host.id, local_nodeid);
return NULL;
} else if (msg->host.size != 0 && safe_str_neq(msg->host.uname, local_name)) {
/* Not for us */
crm_trace("Not for us: %s != %s", msg->host.uname, local_name);
return NULL;
}
msg->sender.id = nodeid;
if (msg->sender.size == 0) {
crm_node_t *peer = crm_get_peer(nodeid, NULL);
if (peer == NULL) {
crm_err("Peer with nodeid=%u is unknown", nodeid);
} else if (peer->uname == NULL) {
crm_err("No uname for peer with nodeid=%u", nodeid);
} else {
crm_notice("Fixing uname for peer with nodeid=%u", nodeid);
msg->sender.size = strlen(peer->uname);
memset(msg->sender.uname, 0, MAX_NAME);
memcpy(msg->sender.uname, peer->uname, msg->sender.size);
}
}
}
crm_trace("Got new%s message (size=%d, %d, %d)",
msg->is_compressed ? " compressed" : "",
ais_data_len(msg), msg->size, msg->compressed_size);
if (kind != NULL) {
*kind = msg->header.id;
}
if (from != NULL) {
*from = msg->sender.uname;
}
if (msg->is_compressed && msg->size > 0) {
int rc = BZ_OK;
char *uncompressed = NULL;
unsigned int new_size = msg->size + 1;
if (check_message_sanity(msg, NULL) == FALSE) {
goto badmsg;
}
crm_trace("Decompressing message data");
uncompressed = calloc(1, new_size);
rc = BZ2_bzBuffToBuffDecompress(uncompressed, &new_size, msg->data, msg->compressed_size, 1, 0);
if (rc != BZ_OK) {
crm_err("Decompression failed: %d", rc);
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_cluster_text(crm_class_cluster, pid_s, TRUE, NULL, crm_msg_ais);
free(pid_s);
return NULL;
} else {
data = strdup(msg->data);
}
if (msg->header.id != crm_class_members) {
/* Is this even needed anymore? */
crm_get_peer(msg->sender.id, msg->sender.uname);
}
if (msg->header.id == crm_class_rmpeer) {
uint32_t id = crm_int_helper(data, NULL);
crm_info("Removing peer %s/%u", data, id);
reap_crm_member(id, NULL);
free(data);
return NULL;
#if SUPPORT_PLUGIN
} else if (is_classic_ais_cluster()) {
plugin_handle_membership(msg);
#endif
}
crm_trace("Payload: %.200s", data);
return data;
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);
free(data);
return NULL;
}
void
pcmk_cpg_membership(cpg_handle_t handle,
const struct cpg_name *groupName,
const struct cpg_address *member_list, size_t member_list_entries,
const struct cpg_address *left_list, size_t left_list_entries,
const struct cpg_address *joined_list, size_t joined_list_entries)
{
int i;
gboolean found = FALSE;
static int counter = 0;
uint32_t local_nodeid = get_local_nodeid(handle);
for (i = 0; i < left_list_entries; i++) {
crm_node_t *peer = crm_get_peer(left_list[i].nodeid, NULL);
crm_info("Left[%d.%d] %s.%u ", counter, i, groupName->value, left_list[i].nodeid);
crm_update_peer_proc(__FUNCTION__, peer, crm_proc_cpg, OFFLINESTATUS);
}
for (i = 0; i < joined_list_entries; i++) {
crm_info("Joined[%d.%d] %s.%u ", counter, i, groupName->value, joined_list[i].nodeid);
}
for (i = 0; i < member_list_entries; i++) {
crm_node_t *peer = crm_get_peer(member_list[i].nodeid, NULL);
crm_info("Member[%d.%d] %s.%u ", counter, i, groupName->value, member_list[i].nodeid);
/* Anyone that is sending us CPG messages must also be a _CPG_ member.
* But its _not_ safe to assume its in the quorum membership.
* We may have just found out its dead and are processing the last couple of messages it sent
*/
crm_update_peer_proc(__FUNCTION__, peer, crm_proc_cpg, ONLINESTATUS);
if(peer && peer->state && crm_is_peer_active(peer) == FALSE) {
time_t now = time(NULL);
/* Co-opt the otherwise unused votes field */
if(peer->votes == 0) {
peer->votes = now;
} else if(now > (60 + peer->votes)) {
/* On the otherhand, if we're still getting messages, at a certain point
* we need to acknowledge our internal cache is probably wrong
*
* Set the threshold to 1 minute
*/
crm_err("Node %s[%u] appears to be online even though we think it is dead", peer->uname, peer->id);
crm_update_peer_state(__FUNCTION__, peer, CRM_NODE_MEMBER, 0);
peer->votes = 0;
}
}
if (local_nodeid == member_list[i].nodeid) {
found = TRUE;
}
}
if (!found) {
crm_err("We're not part of CPG group '%s' anymore!", groupName->value);
cpg_evicted = TRUE;
}
counter++;
}
gboolean
cluster_connect_cpg(crm_cluster_t *cluster)
{
int rc = -1;
int fd = 0;
int retries = 0;
uint32_t id = 0;
crm_node_t *peer = NULL;
cpg_handle_t handle = 0;
struct mainloop_fd_callbacks cpg_fd_callbacks = {
.dispatch = pcmk_cpg_dispatch,
.destroy = cluster->destroy,
};
cpg_callbacks_t cpg_callbacks = {
.cpg_deliver_fn = cluster->cpg.cpg_deliver_fn,
.cpg_confchg_fn = cluster->cpg.cpg_confchg_fn,
/* .cpg_deliver_fn = pcmk_cpg_deliver, */
/* .cpg_confchg_fn = pcmk_cpg_membership, */
};
cpg_evicted = FALSE;
cluster->group.length = 0;
cluster->group.value[0] = 0;
strncpy(cluster->group.value, crm_system_name, 128);
cluster->group.length = strlen(crm_system_name) + 1;
cs_repeat(retries, 30, rc = cpg_initialize(&handle, &cpg_callbacks));
if (rc != CS_OK) {
crm_err("Could not connect to the Cluster Process Group API: %d\n", rc);
goto bail;
}
id = get_local_nodeid(handle);
if (id == 0) {
crm_err("Could not get local node id from the CPG API");
goto bail;
}
cluster->nodeid = id;
retries = 0;
cs_repeat(retries, 30, rc = cpg_join(handle, &cluster->group));
if (rc != CS_OK) {
crm_err("Could not join the CPG group '%s': %d", crm_system_name, rc);
goto bail;
}
rc = cpg_fd_get(handle, &fd);
if (rc != CS_OK) {
crm_err("Could not obtain the CPG API connection: %d\n", rc);
goto bail;
}
pcmk_cpg_handle = handle;
cluster->cpg_handle = handle;
mainloop_add_fd("corosync-cpg", G_PRIORITY_MEDIUM, fd, cluster, &cpg_fd_callbacks);
bail:
if (rc != CS_OK) {
cpg_finalize(handle);
return FALSE;
}
peer = crm_get_peer(id, NULL);
crm_update_peer_proc(__FUNCTION__, peer, crm_proc_cpg, ONLINESTATUS);
return TRUE;
}
gboolean
send_cluster_message_cs(xmlNode * msg, gboolean local, crm_node_t * node, enum crm_ais_msg_types dest)
{
gboolean rc = TRUE;
char *data = NULL;
data = dump_xml_unformatted(msg);
rc = send_cluster_text(crm_class_cluster, data, local, node, dest);
free(data);
return rc;
}
gboolean
send_cluster_text(int class, const char *data,
gboolean local, crm_node_t * node, enum crm_ais_msg_types dest)
{
static int msg_id = 0;
static int local_pid = 0;
static int local_name_len = 0;
static const char *local_name = NULL;
char *target = NULL;
struct iovec *iov;
AIS_Message *msg = NULL;
enum crm_ais_msg_types sender = text2msg_type(crm_system_name);
/* There are only 6 handlers registered to crm_lib_service in plugin.c */
CRM_CHECK(class < 6, crm_err("Invalid message class: %d", class);
return FALSE);
#if !SUPPORT_PLUGIN
CRM_CHECK(dest != crm_msg_ais, return FALSE);
#endif
if(local_name == NULL) {
local_name = get_local_node_name();
}
if(local_name_len == 0 && local_name) {
local_name_len = strlen(local_name);
}
if (data == NULL) {
data = "";
}
if (local_pid == 0) {
local_pid = getpid();
}
if (sender == crm_msg_none) {
sender = local_pid;
}
msg = calloc(1, sizeof(AIS_Message));
msg_id++;
msg->id = msg_id;
msg->header.id = class;
msg->header.error = CS_OK;
msg->host.type = dest;
msg->host.local = local;
if (node) {
if (node->uname) {
target = strdup(node->uname);
msg->host.size = strlen(node->uname);
memset(msg->host.uname, 0, MAX_NAME);
memcpy(msg->host.uname, node->uname, msg->host.size);
} else {
target = g_strdup_printf("%u", node->id);
}
msg->host.id = node->id;
} else {
target = strdup("all");
}
msg->sender.id = 0;
msg->sender.type = sender;
msg->sender.pid = local_pid;
msg->sender.size = local_name_len;
memset(msg->sender.uname, 0, MAX_NAME);
memcpy(msg->sender.uname, local_name, msg->sender.size);
msg->size = 1 + strlen(data);
msg->header.size = sizeof(AIS_Message) + msg->size;
if (msg->size < CRM_BZ2_THRESHOLD) {
msg = realloc(msg, msg->header.size);
memcpy(msg->data, data, msg->size);
} else {
char *compressed = NULL;
unsigned int new_size = 0;
char *uncompressed = strdup(data);
if (crm_compress_string(uncompressed, msg->size, 0, &compressed, &new_size)) {
msg->header.size = sizeof(AIS_Message) + new_size + 1;
msg = realloc(msg, msg->header.size);
memcpy(msg->data, compressed, new_size);
msg->data[new_size] = 0;
msg->is_compressed = TRUE;
msg->compressed_size = new_size;
} else {
msg = realloc(msg, msg->header.size);
memcpy(msg->data, data, msg->size);
}
free(uncompressed);
free(compressed);
}
iov = calloc(1, sizeof(struct iovec));
iov->iov_base = msg;
iov->iov_len = msg->header.size;
if (msg->compressed_size) {
crm_trace("Queueing CPG message %u to %s (%d bytes, %d bytes compressed payload): %.200s",
msg->id, target, iov->iov_len, msg->compressed_size, data);
} else {
crm_trace("Queueing CPG message %u to %s (%d bytes, %d bytes payload): %.200s",
msg->id, target, iov->iov_len, msg->size, data);
}
+ free(target);
#if SUPPORT_PLUGIN
/* The plugin is the only time we dont use CPG messaging */
if(get_cluster_type() == pcmk_cluster_classic_ais) {
return send_plugin_text(class, iov);
}
#endif
send_cpg_iov(iov);
- free(target);
return TRUE;
}
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, "stonith-ng")) {
type = crm_msg_stonith_ng;
} else if (safe_str_eq(text, "attrd")) {
type = crm_msg_attrd;
} else {
/* This will normally be a transient client rather than
* a cluster daemon. Set the type to the pid of the client
*/
int scan_rc = sscanf(text, "%d", &type);
if (scan_rc != 1) {
/* Ensure its sane */
type = crm_msg_none;
}
}
return type;
}
diff --git a/lib/cluster/legacy.c b/lib/cluster/legacy.c
index cb08f51b1f..ed5e18c6a9 100644
--- a/lib/cluster/legacy.c
+++ b/lib/cluster/legacy.c
@@ -1,950 +1,945 @@
/*
* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser 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 library 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
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <crm/cluster/internal.h>
#include <bzlib.h>
#include <crm/common/ipc.h>
#include <crm/cluster.h>
#include <crm/common/mainloop.h>
#include <sys/utsname.h>
#include <sys/socket.h>
#include <netdb.h>
#if SUPPORT_COROSYNC
# include <corosync/confdb.h>
# include <corosync/corodefs.h>
# include <corosync/cpg.h>
# include <corosync/cfg.h>
#endif
#if HAVE_CMAP
# include <corosync/cmap.h>
#endif
#if SUPPORT_CMAN
# include <libcman.h>
cman_handle_t pcmk_cman_handle = NULL;
#endif
int ais_membership_timer = 0;
gboolean ais_membership_force = FALSE;
int plugin_dispatch(gpointer user_data);
int ais_fd_sync = -1;
int ais_fd_async = -1; /* never send messages via this channel */
void *ais_ipc_ctx = NULL;
hdb_handle_t ais_ipc_handle = 0;
static gboolean
get_ais_details(uint32_t * id, char **uname)
{
struct iovec iov;
int retries = 0;
int rc = CS_OK;
cs_ipc_header_response_t header;
struct crm_ais_nodeid_resp_s answer;
static uint32_t local_id = 0;
static char *local_uname = NULL;
if(local_id) {
if(id) *id = local_id;
if(uname) *uname = strdup(local_uname);
return TRUE;
}
header.error = CS_OK;
header.id = crm_class_nodeid;
header.size = sizeof(cs_ipc_header_response_t);
iov.iov_base = &header;
iov.iov_len = header.size;
retry:
errno = 0;
rc = coroipcc_msg_send_reply_receive(ais_ipc_handle, &iov, 1, &answer, sizeof(answer));
if (rc == CS_OK) {
CRM_CHECK(answer.header.size == sizeof(struct crm_ais_nodeid_resp_s),
crm_err("Odd message: id=%d, size=%d, error=%d",
answer.header.id, answer.header.size, answer.header.error));
CRM_CHECK(answer.header.id == crm_class_nodeid,
crm_err("Bad response id: %d", answer.header.id));
}
if ((rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) && retries < 20) {
retries++;
crm_info("Peer overloaded: Re-sending message (Attempt %d of 20)", retries);
sleep(retries); /* Proportional back off */
goto retry;
}
if (rc != CS_OK) {
crm_err("Sending nodeid request: FAILED (rc=%d): %s", rc, ais_error2text(rc));
return FALSE;
} else if (answer.header.error != CS_OK) {
crm_err("Bad response from peer: (rc=%d): %s", rc, ais_error2text(rc));
return FALSE;
}
crm_info("Server details: id=%u uname=%s cname=%s", answer.id, answer.uname, answer.cname);
local_id = answer.id;
local_uname = strdup(answer.uname);
if(id) *id = local_id;
if(uname) *uname = strdup(local_uname);
return TRUE;
}
bool
send_plugin_text(int class, struct iovec *iov)
{
int rc = CS_OK;
int retries = 0;
int buf_len = sizeof(cs_ipc_header_response_t);
char *buf = malloc(buf_len);
AIS_Message *ais_msg = (AIS_Message*)iov[0].iov_base;
cs_ipc_header_response_t *header = (cs_ipc_header_response_t *) buf;
/* There are only 6 handlers registered to crm_lib_service in plugin.c */
CRM_CHECK(class < 6, crm_err("Invalid message class: %d", class);
return FALSE);
do {
if (rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) {
retries++;
crm_info("Peer overloaded or membership in flux:"
" Re-sending message (Attempt %d of 20)", retries);
sleep(retries); /* Proportional back off */
}
errno = 0;
rc = coroipcc_msg_send_reply_receive(ais_ipc_handle, iov, 1, buf, buf_len);
} while ((rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) && retries < 20);
if (rc == CS_OK) {
CRM_CHECK(header->size == sizeof(cs_ipc_header_response_t),
crm_err("Odd message: id=%d, size=%d, class=%d, error=%d",
header->id, header->size, class, header->error));
CRM_ASSERT(buf_len >= header->size);
CRM_CHECK(header->id == CRM_MESSAGE_IPC_ACK,
crm_err("Bad response id (%d) for request (%d)", header->id,
ais_msg->header.id));
CRM_CHECK(header->error == CS_OK, rc = header->error);
} else {
crm_perror(LOG_ERR, "Sending plugin message %d FAILED: %s (%d)",
ais_msg->id, ais_error2text(rc), rc);
}
free(iov[0].iov_base);
free(iov);
free(buf);
return (rc == CS_OK);
}
void
terminate_cs_connection(crm_cluster_t *cluster)
{
crm_notice("Disconnecting from Corosync");
if (is_classic_ais_cluster()) {
if (ais_ipc_handle) {
crm_trace("Disconnecting plugin");
coroipcc_service_disconnect(ais_ipc_handle);
ais_ipc_handle = 0;
} else {
crm_info("No plugin connection");
}
}
cluster_disconnect_cpg(cluster);
# if SUPPORT_CMAN
if (is_cman_cluster()) {
if (pcmk_cman_handle) {
crm_info("Disconnecting cman");
if (cman_stop_notification(pcmk_cman_handle) >= 0) {
crm_info("Destroying cman");
cman_finish(pcmk_cman_handle);
}
} else {
crm_info("No cman connection");
}
}
# endif
ais_fd_async = -1;
ais_fd_sync = -1;
}
void
plugin_handle_membership(AIS_Message *msg)
{
if (msg->header.id == crm_class_members || msg->header.id == crm_class_quorum) {
xmlNode *member = NULL;
const char *value = NULL;
gboolean quorate = FALSE;
xmlNode *xml = string2xml(msg->data);
if (xml == NULL) {
crm_err("Invalid membership update: %s", msg->data);
return;
}
value = crm_element_value(xml, "quorate");
CRM_CHECK(value != NULL, crm_log_xml_err(xml, "No quorum value:"); return);
if (crm_is_true(value)) {
quorate = TRUE;
}
value = crm_element_value(xml, "id");
CRM_CHECK(value != NULL, crm_log_xml_err(xml, "No membership id"); return);
crm_peer_seq = crm_int_helper(value, NULL);
if (quorate != crm_have_quorum) {
crm_notice("Membership %s: quorum %s", value, quorate ? "acquired" : "lost");
crm_have_quorum = quorate;
} else {
crm_info("Membership %s: quorum %s", value, quorate ? "retained" : "still lost");
}
for (member = __xml_first_child(xml); member != NULL; member = __xml_next(member)) {
const char *id_s = crm_element_value(member, "id");
const char *addr = crm_element_value(member, "addr");
const char *uname = crm_element_value(member, "uname");
const char *state = crm_element_value(member, "state");
const char *born_s = crm_element_value(member, "born");
const char *seen_s = crm_element_value(member, "seen");
const char *votes_s = crm_element_value(member, "votes");
const char *procs_s = crm_element_value(member, "processes");
int votes = crm_int_helper(votes_s, NULL);
unsigned int id = crm_int_helper(id_s, NULL);
unsigned int procs = crm_int_helper(procs_s, NULL);
/* TODO: These values will contain garbage if version < 0.7.1 */
uint64_t born = crm_int_helper(born_s, NULL);
uint64_t seen = crm_int_helper(seen_s, NULL);
crm_update_peer(__FUNCTION__, id, born, seen, votes, procs, uname, uname, addr, state);
}
free_xml(xml);
}
}
static void
legacy_default_deliver_message(cpg_handle_t handle,
const struct cpg_name *groupName,
uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
{
uint32_t kind = 0;
const char *from = NULL;
char *data = pcmk_message_common_cs(handle, nodeid, pid, msg, &kind, &from);
free(data);
}
int
plugin_dispatch(gpointer user_data)
{
int rc = CS_OK;
crm_cluster_t *cluster = (crm_cluster_t *) user_data;
do {
char *buffer = NULL;
rc = coroipcc_dispatch_get(ais_ipc_handle, (void **)&buffer, 0);
if (rc == CS_ERR_TRY_AGAIN || rc == CS_ERR_QUEUE_FULL) {
return 0;
}
if (rc != CS_OK) {
crm_perror(LOG_ERR, "Receiving message body failed: (%d) %s", rc, ais_error2text(rc));
return -1;
}
if (buffer == NULL) {
/* NULL is a legal "no message afterall" value */
return 0;
}
/*
cpg_deliver_fn_t(cpg_handle_t handle, const struct cpg_name *group_name,
uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len);
*/
if (cluster && cluster->cpg.cpg_deliver_fn) {
cluster->cpg.cpg_deliver_fn(0, NULL, 0, 0, buffer, 0);
} else {
legacy_default_deliver_message(0, NULL, 0, 0, buffer, 0);
}
coroipcc_dispatch_put(ais_ipc_handle);
} while (ais_ipc_handle);
return 0;
}
static void
plugin_destroy(gpointer user_data)
{
crm_err("AIS connection terminated");
ais_fd_sync = -1;
crm_exit(ENOTCONN);
}
# if SUPPORT_CMAN
static int
pcmk_cman_dispatch(gpointer user_data)
{
int rc = cman_dispatch(pcmk_cman_handle, CMAN_DISPATCH_ALL);
if (rc < 0) {
crm_err("Connection to cman failed: %d", rc);
pcmk_cman_handle = 0;
return FALSE;
}
return TRUE;
}
# define MAX_NODES 256
static void
cman_event_callback(cman_handle_t handle, void *privdata, int reason, int arg)
{
int rc = 0, lpc = 0, node_count = 0;
cman_cluster_t cluster;
static cman_node_t cman_nodes[MAX_NODES];
gboolean(*dispatch) (unsigned long long, gboolean) = privdata;
switch (reason) {
case CMAN_REASON_STATECHANGE:
memset(&cluster, 0, sizeof(cluster));
rc = cman_get_cluster(pcmk_cman_handle, &cluster);
if (rc < 0) {
crm_err("Couldn't query cman cluster details: %d %d", rc, errno);
return;
}
crm_peer_seq = cluster.ci_generation;
if (arg != crm_have_quorum) {
crm_notice("Membership %llu: quorum %s", crm_peer_seq, arg ? "acquired" : "lost");
crm_have_quorum = arg;
} else {
crm_info("Membership %llu: quorum %s", crm_peer_seq,
arg ? "retained" : "still lost");
}
rc = cman_get_nodes(pcmk_cman_handle, MAX_NODES, &node_count, cman_nodes);
if (rc < 0) {
crm_err("Couldn't query cman node list: %d %d", rc, errno);
return;
}
for (lpc = 0; lpc < node_count; lpc++) {
if (cman_nodes[lpc].cn_nodeid == 0) {
/* Never allow node ID 0 to be considered a member #315711 */
/* Skip entirely, its a qdisk */
continue;
}
crm_update_peer(__FUNCTION__, cman_nodes[lpc].cn_nodeid,
cman_nodes[lpc].cn_incarnation,
cman_nodes[lpc].cn_member ? crm_peer_seq : 0, 0, 0,
cman_nodes[lpc].cn_name, cman_nodes[lpc].cn_name, NULL,
cman_nodes[lpc].cn_member ? CRM_NODE_MEMBER : CRM_NODE_LOST);
}
if (dispatch) {
dispatch(crm_peer_seq, crm_have_quorum);
}
break;
case CMAN_REASON_TRY_SHUTDOWN:
/* Always reply with a negative - pacemaker needs to be stopped first */
crm_notice("CMAN wants to shut down: %s", arg ? "forced" : "optional");
cman_replyto_shutdown(pcmk_cman_handle, 0);
break;
case CMAN_REASON_CONFIG_UPDATE:
/* Ignore */
break;
}
}
# endif
gboolean
init_cman_connection(gboolean(*dispatch) (unsigned long long, gboolean), void (*destroy) (gpointer))
{
# if SUPPORT_CMAN
int rc = -1, fd = -1;
cman_cluster_t cluster;
struct mainloop_fd_callbacks cman_fd_callbacks = {
.dispatch = pcmk_cman_dispatch,
.destroy = destroy,
};
crm_info("Configuring Pacemaker to obtain quorum from cman");
memset(&cluster, 0, sizeof(cluster));
pcmk_cman_handle = cman_init(dispatch);
if (pcmk_cman_handle == NULL || cman_is_active(pcmk_cman_handle) == FALSE) {
crm_err("Couldn't connect to cman");
goto cman_bail;
}
rc = cman_start_notification(pcmk_cman_handle, cman_event_callback);
if (rc < 0) {
crm_err("Couldn't register for cman notifications: %d %d", rc, errno);
goto cman_bail;
}
/* Get the current membership state */
cman_event_callback(pcmk_cman_handle, dispatch, CMAN_REASON_STATECHANGE,
cman_is_quorate(pcmk_cman_handle));
fd = cman_get_fd(pcmk_cman_handle);
mainloop_add_fd("cman", G_PRIORITY_MEDIUM, fd, dispatch, &cman_fd_callbacks);
cman_bail:
if (rc < 0) {
cman_finish(pcmk_cman_handle);
return FALSE;
}
# else
crm_err("cman qorum is not supported in this build");
crm_exit(DAEMON_RESPAWN_STOP);
# endif
return TRUE;
}
# ifdef SUPPORT_COROSYNC
gboolean
init_quorum_connection(gboolean(*dispatch) (unsigned long long, gboolean),
void (*destroy) (gpointer))
{
crm_err("The Corosync quorum API is not supported in this build");
crm_exit(DAEMON_RESPAWN_STOP);
return TRUE;
}
static gboolean
init_cs_connection_classic(crm_cluster_t * cluster)
{
int rc;
int pid = 0;
char *pid_s = NULL;
const char *name = NULL;
crm_node_t *peer = NULL;
enum crm_proc_flag proc = 0;
struct mainloop_fd_callbacks ais_fd_callbacks = {
.dispatch = plugin_dispatch,
.destroy = cluster->destroy,
};
crm_info("Creating connection to our Corosync plugin");
rc = coroipcc_service_connect(COROSYNC_SOCKET_NAME, PCMK_SERVICE_ID,
AIS_IPC_MESSAGE_SIZE, AIS_IPC_MESSAGE_SIZE, AIS_IPC_MESSAGE_SIZE,
&ais_ipc_handle);
if (ais_ipc_handle) {
coroipcc_fd_get(ais_ipc_handle, &ais_fd_async);
} else {
crm_info("Connection to our Corosync plugin (%d) failed: %s (%d)",
PCMK_SERVICE_ID, strerror(errno), errno);
return FALSE;
}
if (ais_fd_async <= 0 && rc == CS_OK) {
crm_err("No context created, but connection reported 'ok'");
rc = CS_ERR_LIBRARY;
}
if (rc != CS_OK) {
crm_info("Connection to our Corosync plugin (%d) failed: %s (%d)", PCMK_SERVICE_ID,
ais_error2text(rc), rc);
}
if (rc != CS_OK) {
return FALSE;
}
if (ais_fd_callbacks.destroy == NULL) {
ais_fd_callbacks.destroy = plugin_destroy;
}
mainloop_add_fd("corosync-plugin", G_PRIORITY_MEDIUM, ais_fd_async, cluster, &ais_fd_callbacks);
crm_info("AIS connection established");
pid = getpid();
pid_s = crm_itoa(pid);
send_cluster_text(crm_class_cluster, pid_s, TRUE, NULL, crm_msg_ais);
free(pid_s);
cluster->nodeid = get_local_nodeid(0);
name = get_local_node_name();
get_ais_details(NULL, &(cluster->uname));
if (safe_str_neq(name, cluster->uname)) {
crm_crit("Node name mismatch! Corosync supplied %s but our lookup returned %s",
cluster->uname, name);
crm_notice
("Node name mismatches usually occur when assigned automatically by DHCP servers");
crm_exit(ENOTUNIQ);
}
proc = text2proc(crm_system_name);
peer = crm_get_peer(cluster->nodeid, cluster->uname);
crm_update_peer_proc(__FUNCTION__, peer, proc|crm_proc_plugin, ONLINESTATUS);
return TRUE;
}
static int
pcmk_mcp_dispatch(const char *buffer, ssize_t length, gpointer userdata)
{
xmlNode *msg = string2xml(buffer);
if (msg && is_classic_ais_cluster()) {
xmlNode *node = NULL;
for (node = __xml_first_child(msg); node != NULL; node = __xml_next(node)) {
int id = 0;
int children = 0;
const char *uname = crm_element_value(node, "uname");
crm_element_value_int(node, "id", &id);
crm_element_value_int(node, "processes", &children);
if (id == 0) {
crm_log_xml_err(msg, "Bad Update");
} else {
crm_node_t *peer = crm_get_peer(id, uname);
crm_update_peer_proc(__FUNCTION__, peer, children, NULL);
}
}
}
free_xml(msg);
return 0;
}
static void
pcmk_mcp_destroy(gpointer user_data)
{
void (*callback) (gpointer data) = user_data;
if (callback) {
callback(NULL);
}
}
gboolean
init_cs_connection(crm_cluster_t * cluster)
{
int retries = 0;
static struct ipc_client_callbacks mcp_callbacks = {
.dispatch = pcmk_mcp_dispatch,
.destroy = pcmk_mcp_destroy
};
while (retries < 5) {
int rc = init_cs_connection_once(cluster);
retries++;
switch (rc) {
case CS_OK:
if (getenv("HA_mcp") && get_cluster_type() != pcmk_cluster_cman) {
xmlNode *poke = create_xml_node(NULL, "poke");
mainloop_io_t *ipc =
mainloop_add_ipc_client(CRM_SYSTEM_MCP, G_PRIORITY_MEDIUM, 0,
cluster->destroy, &mcp_callbacks);
crm_ipc_send(mainloop_get_ipc_client(ipc), poke, 0, 0, NULL);
free_xml(poke);
}
return TRUE;
break;
case CS_ERR_TRY_AGAIN:
case CS_ERR_QUEUE_FULL:
sleep(retries);
break;
default:
return FALSE;
}
}
crm_err("Retry count exceeded: %d", retries);
return FALSE;
}
char *
classic_node_name(uint32_t nodeid)
{
return NULL; /* Always use the uname() default for localhost. No way to look up peers */
}
char *
cman_node_name(uint32_t nodeid)
{
char *name = NULL;
# if SUPPORT_CMAN
cman_node_t us;
cman_handle_t cman;
cman = cman_init(NULL);
if (cman != NULL && cman_is_active(cman)) {
us.cn_name[0] = 0;
cman_get_node(cman, nodeid, &us);
name = strdup(us.cn_name);
crm_info("Using CMAN node name %s for %u", name, nodeid);
}
cman_finish(cman);
# endif
if (name == NULL) {
crm_debug("Unable to get node name for nodeid %u", nodeid);
}
return name;
}
extern int set_cluster_type(enum cluster_type_e type);
gboolean
init_cs_connection_once(crm_cluster_t * cluster)
{
- const char *uuid = NULL;
crm_node_t *peer = NULL;
enum cluster_type_e stack = get_cluster_type();
crm_peer_init();
/* Here we just initialize comms */
switch (stack) {
case pcmk_cluster_classic_ais:
if (init_cs_connection_classic(cluster) == FALSE) {
return FALSE;
}
break;
case pcmk_cluster_cman:
if (cluster_connect_cpg(cluster) == FALSE) {
return FALSE;
}
cluster->uname = cman_node_name(0 /* CMAN_NODEID_US */ );
break;
case pcmk_cluster_heartbeat:
crm_info("Could not find an active corosync based cluster");
return FALSE;
break;
default:
crm_err("Invalid cluster type: %s (%d)", name_for_cluster_type(stack), stack);
return FALSE;
break;
}
crm_info("Connection to '%s': established", name_for_cluster_type(stack));
cluster->nodeid = get_local_nodeid(0);
if(cluster->nodeid == 0) {
crm_err("Could not establish local nodeid");
return FALSE;
}
cluster->uname = get_node_name(0);
if(cluster->uname == NULL) {
crm_err("Could not establish local node name");
return FALSE;
}
/* Ensure the local node always exists */
peer = crm_get_peer(cluster->nodeid, cluster->uname);
- uuid = get_corosync_uuid(peer);
-
- if(uuid) {
- cluster->uuid = strdup(uuid);
- }
+ cluster->uuid = get_corosync_uuid(peer);
return TRUE;
}
gboolean
check_message_sanity(const AIS_Message * msg, const 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 != CS_OK) {
crm_warn("Message header contains an error: %d", msg->header.error);
sane = FALSE;
}
if (sane && ais_data_len(msg) != tmp_size) {
crm_warn("Message payload size is incorrect: expected %d, got %d", ais_data_len(msg),
tmp_size);
sane = TRUE;
}
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_trace
("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
static int
get_config_opt(confdb_handle_t config,
hdb_handle_t object_handle, const char *key, char **value, const char *fallback)
{
size_t len = 0;
char *env_key = NULL;
const char *env_value = NULL;
char buffer[256];
if (*value) {
free(*value);
*value = NULL;
}
if (object_handle > 0) {
if (CS_OK == confdb_key_get(config, object_handle, key, strlen(key), &buffer, &len)) {
*value = strdup(buffer);
}
}
if (*value) {
crm_info("Found '%s' for option: %s", *value, key);
return 0;
}
env_key = crm_concat("HA", key, '_');
env_value = getenv(env_key);
free(env_key);
if (*value) {
crm_info("Found '%s' in ENV for option: %s", *value, key);
*value = strdup(env_value);
return 0;
}
if (fallback) {
crm_info("Defaulting to '%s' for option: %s", fallback, key);
*value = strdup(fallback);
} else {
crm_info("No default for option: %s", key);
}
return -1;
}
static confdb_handle_t
config_find_init(confdb_handle_t config)
{
cs_error_t rc = CS_OK;
confdb_handle_t local_handle = OBJECT_PARENT_HANDLE;
rc = confdb_object_find_start(config, local_handle);
if (rc == CS_OK) {
return local_handle;
} else {
crm_err("Couldn't create search context: %d", rc);
}
return 0;
}
static hdb_handle_t
config_find_next(confdb_handle_t config, const char *name, confdb_handle_t top_handle)
{
cs_error_t rc = CS_OK;
hdb_handle_t local_handle = 0;
if (top_handle == 0) {
crm_err("Couldn't search for %s: no valid context", name);
return 0;
}
crm_trace("Searching for %s in " HDB_X_FORMAT, name, top_handle);
rc = confdb_object_find(config, top_handle, name, strlen(name), &local_handle);
if (rc != CS_OK) {
crm_info("No additional configuration supplied for: %s", name);
local_handle = 0;
} else {
crm_info("Processing additional %s options...", name);
}
return local_handle;
}
enum cluster_type_e
find_corosync_variant(void)
{
confdb_handle_t config;
enum cluster_type_e found = pcmk_cluster_unknown;
int rc;
char *value = NULL;
confdb_handle_t top_handle = 0;
hdb_handle_t local_handle = 0;
static confdb_callbacks_t callbacks = { };
rc = confdb_initialize(&config, &callbacks);
if (rc != CS_OK) {
crm_debug("Could not initialize Cluster Configuration Database API instance error %d", rc);
return found;
}
top_handle = config_find_init(config);
local_handle = config_find_next(config, "service", top_handle);
while (local_handle) {
get_config_opt(config, local_handle, "name", &value, NULL);
if (safe_str_eq("pacemaker", value)) {
found = pcmk_cluster_classic_ais;
get_config_opt(config, local_handle, "ver", &value, "0");
crm_trace("Found Pacemaker plugin version: %s", value);
break;
}
local_handle = config_find_next(config, "service", top_handle);
}
if (found == pcmk_cluster_unknown) {
top_handle = config_find_init(config);
local_handle = config_find_next(config, "quorum", top_handle);
get_config_opt(config, local_handle, "provider", &value, NULL);
if (safe_str_eq("quorum_cman", value)) {
crm_trace("Found CMAN quorum provider");
found = pcmk_cluster_cman;
}
}
free(value);
confdb_finalize(config);
if (found == pcmk_cluster_unknown) {
crm_err
("Corosync is running, but Pacemaker could not find the CMAN or Pacemaker plugin loaded");
found = pcmk_cluster_invalid;
}
return found;
}
gboolean
crm_is_corosync_peer_active(const crm_node_t * node)
{
enum crm_proc_flag proc = crm_proc_none;
if (node == NULL) {
crm_trace("NULL");
return FALSE;
} else if (safe_str_neq(node->state, CRM_NODE_MEMBER)) {
crm_trace("%s: state=%s", node->uname, node->state);
return FALSE;
} else if (is_cman_cluster() && (node->processes & crm_proc_cpg)) {
/* If we can still talk to our peer process on that node,
* then its also part of the corosync membership
*/
crm_trace("%s: processes=%.8x", node->uname, node->processes);
return TRUE;
} else if (is_classic_ais_cluster()) {
if (node->processes < crm_proc_none) {
crm_debug("%s: unknown process list, assuming active for now", node->uname);
return TRUE;
} else if (is_set(node->processes, crm_proc_none)) {
crm_debug("%s: all processes are inactive", node->uname);
return FALSE;
} else if (is_not_set(node->processes, crm_proc_plugin)) {
crm_trace("%s: processes=%.8x", node->uname, node->processes);
return FALSE;
}
}
proc = text2proc(crm_system_name);
if (proc > crm_proc_none && (node->processes & proc) == 0) {
crm_trace("%s: proc %.8x not in %.8x", node->uname, proc, node->processes);
return FALSE;
}
return TRUE;
}