Page MenuHomeClusterLabs Projects
Files F4624498

No OneTemporary
Actions

View Options

This file is larger than 256 KB, so syntax highlighting was skipped.
diff --git a/attrd/commands.c b/attrd/commands.c
index 901919a115..442c5f89d1 100644
--- a/attrd/commands.c
+++ b/attrd/commands.c
@@ -1,1027 +1,1023 @@
/*
* Copyright (C) 2013 Andrew Beekhof <andrew@beekhof.net>
*
* 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 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <sys/types.h>
#include <regex.h>
#include <glib.h>
#include <crm/msg_xml.h>
#include <crm/cluster.h>
#include <crm/cib.h>
#include <crm/cluster/internal.h>
#include <crm/cluster/election.h>
#include <crm/cib/internal.h>
#include <internal.h>
#define ATTRD_PROTOCOL_VERSION "1"
int last_cib_op_done = 0;
char *peer_writer = NULL;
GHashTable *attributes = NULL;
typedef struct attribute_s {
char *uuid; /* TODO: Remove if at all possible */
char *id;
char *set;
GHashTable *values;
int update;
int timeout_ms;
/* TODO: refactor these three as a bitmask */
bool changed; /* whether attribute value has changed since last write */
bool unknown_peer_uuids; /* whether we know we're missing a peer uuid */
gboolean is_private; /* whether to keep this attribute out of the CIB */
mainloop_timer_t *timer;
char *user;
} attribute_t;
typedef struct attribute_value_s {
uint32_t nodeid;
gboolean is_remote;
char *nodename;
char *current;
char *requested;
char *stored;
} attribute_value_t;
void write_attribute(attribute_t *a);
void write_or_elect_attribute(attribute_t *a);
void attrd_peer_update(crm_node_t *peer, xmlNode *xml, const char *host, bool filter);
void attrd_peer_sync(crm_node_t *peer, xmlNode *xml);
void attrd_peer_remove(uint32_t nodeid, const char *host, gboolean uncache, const char *source);
static gboolean
send_attrd_message(crm_node_t * node, xmlNode * data)
{
crm_xml_add(data, F_TYPE, T_ATTRD);
crm_xml_add(data, F_ATTRD_IGNORE_LOCALLY, "atomic-version"); /* Tell older versions to ignore our messages */
crm_xml_add(data, F_ATTRD_VERSION, ATTRD_PROTOCOL_VERSION);
crm_xml_add_int(data, F_ATTRD_WRITER, election_state(writer));
return send_cluster_message(node, crm_msg_attrd, data, TRUE);
}
static gboolean
attribute_timer_cb(gpointer data)
{
attribute_t *a = data;
crm_trace("Dampen interval expired for %s in state %d", a->id, election_state(writer));
write_or_elect_attribute(a);
return FALSE;
}
static void
free_attribute_value(gpointer data)
{
attribute_value_t *v = data;
free(v->nodename);
free(v->current);
free(v->requested);
free(v->stored);
free(v);
}
void
free_attribute(gpointer data)
{
attribute_t *a = data;
if(a) {
free(a->id);
free(a->set);
free(a->uuid);
free(a->user);
mainloop_timer_del(a->timer);
g_hash_table_destroy(a->values);
free(a);
}
}
static xmlNode *
build_attribute_xml(
xmlNode *parent, const char *name, const char *set, const char *uuid, unsigned int timeout_ms, const char *user,
gboolean is_private, const char *peer, uint32_t peerid, const char *value)
{
xmlNode *xml = create_xml_node(parent, __FUNCTION__);
crm_xml_add(xml, F_ATTRD_ATTRIBUTE, name);
crm_xml_add(xml, F_ATTRD_SET, set);
crm_xml_add(xml, F_ATTRD_KEY, uuid);
crm_xml_add(xml, F_ATTRD_USER, user);
crm_xml_add(xml, F_ATTRD_HOST, peer);
crm_xml_add_int(xml, F_ATTRD_HOST_ID, peerid);
crm_xml_add(xml, F_ATTRD_VALUE, value);
crm_xml_add_int(xml, F_ATTRD_DAMPEN, timeout_ms/1000);
crm_xml_add_int(xml, F_ATTRD_IS_PRIVATE, is_private);
return xml;
}
static attribute_t *
create_attribute(xmlNode *xml)
{
int dampen = 0;
const char *value = crm_element_value(xml, F_ATTRD_DAMPEN);
attribute_t *a = calloc(1, sizeof(attribute_t));
a->id = crm_element_value_copy(xml, F_ATTRD_ATTRIBUTE);
a->set = crm_element_value_copy(xml, F_ATTRD_SET);
a->uuid = crm_element_value_copy(xml, F_ATTRD_KEY);
a->values = g_hash_table_new_full(crm_strcase_hash, crm_strcase_equal, NULL, free_attribute_value);
crm_element_value_int(xml, F_ATTRD_IS_PRIVATE, &a->is_private);
#if ENABLE_ACL
crm_trace("Performing all %s operations as user '%s'", a->id, a->user);
a->user = crm_element_value_copy(xml, F_ATTRD_USER);
#endif
if(value) {
dampen = crm_get_msec(value);
crm_trace("Created attribute %s with delay %dms (%s)", a->id, dampen, value);
} else {
crm_trace("Created attribute %s with no delay", a->id);
}
if(dampen > 0) {
a->timeout_ms = dampen;
a->timer = mainloop_timer_add(a->id, a->timeout_ms, FALSE, attribute_timer_cb, a);
}
g_hash_table_replace(attributes, a->id, a);
return a;
}
/*!
* \internal
* \brief Respond to a client peer-remove request (i.e. propagate to all peers)
*
* \param[in] client_name Name of client that made request (for log messages)
* \param[in] xml Root of request XML
*
* \return void
*/
void
attrd_client_peer_remove(const char *client_name, xmlNode *xml)
{
const char *host = crm_element_value(xml, F_ATTRD_HOST);
if (host) {
crm_info("Client %s is requesting all values for %s be removed",
client_name, host);
send_attrd_message(NULL, xml); /* ends up at attrd_peer_message() */
} else {
crm_info("Ignoring request by client %s to remove all peer values without specifying peer",
client_name);
}
}
/*!
* \internal
* \brief Respond to a client update request
*
* \param[in] xml Root of request XML
*
* \return void
*/
void
attrd_client_update(xmlNode *xml)
{
attribute_t *a = NULL;
attribute_value_t *v = NULL;
char *key = crm_element_value_copy(xml, F_ATTRD_KEY);
char *set = crm_element_value_copy(xml, F_ATTRD_SET);
char *host = crm_element_value_copy(xml, F_ATTRD_HOST);
const char *attr = crm_element_value(xml, F_ATTRD_ATTRIBUTE);
const char *value = crm_element_value(xml, F_ATTRD_VALUE);
const char *regex = crm_element_value(xml, F_ATTRD_REGEX);
/* If a regex was specified, broadcast a message for each match */
if ((attr == NULL) && regex) {
GHashTableIter aIter;
regex_t *r_patt = calloc(1, sizeof(regex_t));
crm_debug("Setting %s to %s", regex, value);
if (regcomp(r_patt, regex, REG_EXTENDED)) {
crm_err("Bad regex '%s' for update", regex);
} else {
g_hash_table_iter_init(&aIter, attributes);
while (g_hash_table_iter_next(&aIter, (gpointer *) & attr, NULL)) {
int status = regexec(r_patt, attr, 0, NULL, 0);
if (status == 0) {
crm_trace("Matched %s with %s", attr, regex);
crm_xml_add(xml, F_ATTRD_ATTRIBUTE, attr);
send_attrd_message(NULL, xml);
}
}
}
free(key);
free(set);
free(host);
regfree(r_patt);
free(r_patt);
return;
}
if (host == NULL) {
crm_trace("Inferring host");
host = strdup(attrd_cluster->uname);
crm_xml_add(xml, F_ATTRD_HOST, host);
crm_xml_add_int(xml, F_ATTRD_HOST_ID, attrd_cluster->nodeid);
}
a = g_hash_table_lookup(attributes, attr);
/* If value was specified using ++ or += notation, expand to real value */
if (value) {
int offset = 1;
int int_value = 0;
static const int plus_plus_len = 5;
if ((strlen(value) >= (plus_plus_len + 2)) && (value[plus_plus_len] == '+')
&& ((value[plus_plus_len + 1] == '+') || (value[plus_plus_len + 1] == '='))) {
if (a) {
v = g_hash_table_lookup(a->values, host);
}
if (v) {
int_value = char2score(v->current);
}
if (value[plus_plus_len + 1] != '+') {
const char *offset_s = value + (plus_plus_len + 2);
offset = char2score(offset_s);
}
int_value += offset;
if (int_value > INFINITY) {
int_value = INFINITY;
}
crm_info("Expanded %s=%s to %d", attr, value, int_value);
crm_xml_add_int(xml, F_ATTRD_VALUE, int_value);
}
}
if ((peer_writer == NULL) && (election_state(writer) != election_in_progress)) {
crm_info("Starting an election to determine the writer");
election_vote(writer);
}
crm_debug("Broadcasting %s[%s] = %s%s", attr, host, value,
((election_state(writer) == election_won)? " (writer)" : ""));
free(key);
free(set);
free(host);
send_attrd_message(NULL, xml); /* ends up at attrd_peer_message() */
}
/*!
* \internal
* \brief Respond to a client refresh request (i.e. write out all attributes)
*
* \return void
*/
void
attrd_client_refresh(void)
{
GHashTableIter iter;
attribute_t *a = NULL;
/* 'refresh' forces a write of the current value of all attributes
* Cancel any existing timers, we're writing it NOW
*/
g_hash_table_iter_init(&iter, attributes);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) & a)) {
mainloop_timer_stop(a->timer);
}
crm_info("Updating all attributes");
write_attributes(TRUE, FALSE);
}
/*!
* \internal
* \brief Build the XML reply to a client query
*
* param[in] attr Name of requested attribute
* param[in] host Name of requested host (or NULL for all hosts)
*
* \return New XML reply
* \note Caller is responsible for freeing the resulting XML
*/
static xmlNode *build_query_reply(const char *attr, const char *host)
{
xmlNode *reply = create_xml_node(NULL, __FUNCTION__);
attribute_t *a;
if (reply == NULL) {
return NULL;
}
crm_xml_add(reply, F_TYPE, T_ATTRD);
crm_xml_add(reply, F_ATTRD_VERSION, ATTRD_PROTOCOL_VERSION);
/* If desired attribute exists, add its value(s) to the reply */
a = g_hash_table_lookup(attributes, attr);
if (a) {
attribute_value_t *v;
xmlNode *host_value;
crm_xml_add(reply, F_ATTRD_ATTRIBUTE, attr);
/* Allow caller to use "localhost" to refer to local node */
if (safe_str_eq(host, "localhost")) {
host = attrd_cluster->uname;
crm_trace("Mapped localhost to %s", host);
}
/* If a specific node was requested, add its value */
if (host) {
v = g_hash_table_lookup(a->values, host);
host_value = create_xml_node(reply, XML_CIB_TAG_NODE);
if (host_value == NULL) {
free_xml(reply);
return NULL;
}
crm_xml_add(host_value, F_ATTRD_HOST, host);
crm_xml_add(host_value, F_ATTRD_VALUE, (v? v->current : NULL));
/* Otherwise, add all nodes' values */
} else {
GHashTableIter iter;
g_hash_table_iter_init(&iter, a->values);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &v)) {
host_value = create_xml_node(reply, XML_CIB_TAG_NODE);
if (host_value == NULL) {
free_xml(reply);
return NULL;
}
crm_xml_add(host_value, F_ATTRD_HOST, v->nodename);
crm_xml_add(host_value, F_ATTRD_VALUE, v->current);
}
}
}
return reply;
}
/*!
* \internal
* \brief Respond to a client query
*
* \param[in] client Who queried us
* \param[in] query Root of query XML
*
* \return void
*/
void
attrd_client_query(crm_client_t *client, uint32_t id, uint32_t flags, xmlNode *query)
{
const char *attr;
const char *origin = crm_element_value(query, F_ORIG);
ssize_t rc;
xmlNode *reply;
if (origin == NULL) {
origin = "unknown client";
}
crm_debug("Query arrived from %s", origin);
/* Request must specify attribute name to query */
attr = crm_element_value(query, F_ATTRD_ATTRIBUTE);
if (attr == NULL) {
crm_warn("Ignoring malformed query from %s (no attribute name given)",
origin);
return;
}
/* Build the XML reply */
reply = build_query_reply(attr, crm_element_value(query, F_ATTRD_HOST));
if (reply == NULL) {
crm_err("Could not respond to query from %s: could not create XML reply",
origin);
return;
}
crm_log_xml_trace(reply, "Reply");
/* Send the reply to the client */
client->request_id = 0;
if ((rc = crm_ipcs_send(client, id, reply, flags)) < 0) {
crm_err("Could not respond to query from %s: %s (%d)",
origin, pcmk_strerror(-rc), -rc);
}
free_xml(reply);
}
void
attrd_peer_message(crm_node_t *peer, xmlNode *xml)
{
int peer_state = 0;
const char *v = crm_element_value(xml, F_ATTRD_VERSION);
const char *op = crm_element_value(xml, F_ATTRD_TASK);
const char *election_op = crm_element_value(xml, F_CRM_TASK);
const char *host = crm_element_value(xml, F_ATTRD_HOST);
if(election_op) {
enum election_result rc = 0;
crm_xml_add(xml, F_CRM_HOST_FROM, peer->uname);
rc = election_count_vote(writer, xml, TRUE);
switch(rc) {
case election_start:
free(peer_writer);
peer_writer = NULL;
election_vote(writer);
break;
case election_lost:
free(peer_writer);
peer_writer = strdup(peer->uname);
break;
default:
election_check(writer);
break;
}
return;
} else if(v == NULL) {
/* From the non-atomic version */
if (safe_str_eq(op, ATTRD_OP_UPDATE)) {
const char *name = crm_element_value(xml, F_ATTRD_ATTRIBUTE);
crm_trace("Compatibility update of %s from %s", name, peer->uname);
attrd_peer_update(peer, xml, host, FALSE);
} else if (safe_str_eq(op, ATTRD_OP_FLUSH)) {
const char *name = crm_element_value(xml, F_ATTRD_ATTRIBUTE);
attribute_t *a = g_hash_table_lookup(attributes, name);
if(a) {
crm_trace("Compatibility write-out of %s for %s from %s", a->id, op, peer->uname);
write_or_elect_attribute(a);
}
} else if (safe_str_eq(op, ATTRD_OP_REFRESH)) {
GHashTableIter aIter;
attribute_t *a = NULL;
g_hash_table_iter_init(&aIter, attributes);
while (g_hash_table_iter_next(&aIter, NULL, (gpointer *) & a)) {
crm_trace("Compatibility write-out of %s for %s from %s", a->id, op, peer->uname);
write_or_elect_attribute(a);
}
}
}
crm_element_value_int(xml, F_ATTRD_WRITER, &peer_state);
if(election_state(writer) == election_won
&& peer_state == election_won
&& safe_str_neq(peer->uname, attrd_cluster->uname)) {
crm_notice("Detected another attribute writer: %s", peer->uname);
election_vote(writer);
} else if(peer_state == election_won) {
if(peer_writer == NULL) {
peer_writer = strdup(peer->uname);
crm_notice("Recorded attribute writer: %s", peer->uname);
} else if(safe_str_neq(peer->uname, peer_writer)) {
crm_notice("Recorded new attribute writer: %s (was %s)", peer->uname, peer_writer);
free(peer_writer);
peer_writer = strdup(peer->uname);
}
}
if (safe_str_eq(op, ATTRD_OP_UPDATE)) {
attrd_peer_update(peer, xml, host, FALSE);
} else if (safe_str_eq(op, ATTRD_OP_SYNC)) {
attrd_peer_sync(peer, xml);
} else if (safe_str_eq(op, ATTRD_OP_PEER_REMOVE)) {
int host_id = 0;
char *endptr = NULL;
host_id = strtol(host, &endptr, 10);
if (errno != 0 || endptr == host || *endptr != '\0') {
host_id = 0;
} else {
host = NULL;
}
attrd_peer_remove(host_id, host, TRUE, peer->uname);
} else if (safe_str_eq(op, ATTRD_OP_SYNC_RESPONSE)
&& safe_str_neq(peer->uname, attrd_cluster->uname)) {
xmlNode *child = NULL;
crm_notice("Processing %s from %s", op, peer->uname);
for (child = __xml_first_child(xml); child != NULL; child = __xml_next(child)) {
host = crm_element_value(child, F_ATTRD_HOST);
attrd_peer_update(peer, child, host, TRUE);
}
}
}
void
attrd_peer_sync(crm_node_t *peer, xmlNode *xml)
{
GHashTableIter aIter;
GHashTableIter vIter;
attribute_t *a = NULL;
attribute_value_t *v = NULL;
xmlNode *sync = create_xml_node(NULL, __FUNCTION__);
crm_xml_add(sync, F_ATTRD_TASK, ATTRD_OP_SYNC_RESPONSE);
g_hash_table_iter_init(&aIter, attributes);
while (g_hash_table_iter_next(&aIter, NULL, (gpointer *) & a)) {
g_hash_table_iter_init(&vIter, a->values);
while (g_hash_table_iter_next(&vIter, NULL, (gpointer *) & v)) {
crm_debug("Syncing %s[%s] = %s to %s", a->id, v->nodename, v->current, peer?peer->uname:"everyone");
build_attribute_xml(sync, a->id, a->set, a->uuid, a->timeout_ms, a->user, a->is_private,
v->nodename, v->nodeid, v->current);
}
}
crm_debug("Syncing values to %s", peer?peer->uname:"everyone");
send_attrd_message(peer, sync);
free_xml(sync);
}
void
attrd_peer_remove(uint32_t nodeid, const char *host, gboolean uncache, const char *source)
{
attribute_t *a = NULL;
GHashTableIter aIter;
crm_notice("Removing all %s attributes for %s", host, source);
if(host == NULL) {
return;
}
g_hash_table_iter_init(&aIter, attributes);
while (g_hash_table_iter_next(&aIter, NULL, (gpointer *) & a)) {
if(g_hash_table_remove(a->values, host)) {
crm_debug("Removed %s[%s] for %s", a->id, host, source);
}
}
if (uncache) {
crm_remote_peer_cache_remove(host);
reap_crm_member(nodeid, host);
}
}
/*!
* \internal
* \brief Return host's hash table entry (creating one if needed)
*
* \param[in] values Hash table of values
* \param[in] host Name of peer to look up
* \param[in] xml XML describing the attribute
*
* \return Pointer to new or existing hash table entry
*/
static attribute_value_t *
attrd_lookup_or_create_value(GHashTable *values, const char *host, xmlNode *xml)
{
attribute_value_t *v = g_hash_table_lookup(values, host);
if (v == NULL) {
v = calloc(1, sizeof(attribute_value_t));
CRM_ASSERT(v != NULL);
v->nodename = strdup(host);
CRM_ASSERT(v->nodename != NULL);
crm_element_value_int(xml, F_ATTRD_IS_REMOTE, &v->is_remote);
if (v->is_remote == TRUE) {
crm_remote_peer_cache_add(host);
}
g_hash_table_replace(values, v->nodename, v);
}
return(v);
}
void
attrd_peer_update(crm_node_t *peer, xmlNode *xml, const char *host, bool filter)
{
bool changed = FALSE;
attribute_value_t *v = NULL;
const char *attr = crm_element_value(xml, F_ATTRD_ATTRIBUTE);
const char *value = crm_element_value(xml, F_ATTRD_VALUE);
attribute_t *a = g_hash_table_lookup(attributes, attr);
if(a == NULL) {
a = create_attribute(xml);
}
if(host == NULL) {
GHashTableIter vIter;
g_hash_table_iter_init(&vIter, a->values);
crm_debug("Setting %s for all hosts to %s", attr, value);
xml_remove_prop(xml, F_ATTRD_HOST_ID);
while (g_hash_table_iter_next(&vIter, (gpointer *) & host, NULL)) {
attrd_peer_update(peer, xml, host, filter);
}
return;
}
v = attrd_lookup_or_create_value(a->values, host, xml);
if(filter
&& safe_str_neq(v->current, value)
&& safe_str_eq(host, attrd_cluster->uname)) {
xmlNode *sync = create_xml_node(NULL, __FUNCTION__);
crm_notice("%s[%s]: local value '%s' takes priority over '%s' from %s",
a->id, host, v->current, value, peer->uname);
crm_xml_add(sync, F_ATTRD_TASK, ATTRD_OP_SYNC_RESPONSE);
v = g_hash_table_lookup(a->values, host);
build_attribute_xml(sync, a->id, a->set, a->uuid, a->timeout_ms, a->user, a->is_private,
v->nodename, v->nodeid, v->current);
crm_xml_add_int(sync, F_ATTRD_WRITER, election_state(writer));
send_attrd_message(peer, sync);
free_xml(sync);
} else if(safe_str_neq(v->current, value)) {
crm_info("Setting %s[%s]: %s -> %s from %s", attr, host, v->current, value, peer->uname);
free(v->current);
if(value) {
v->current = strdup(value);
} else {
v->current = NULL;
}
changed = TRUE;
} else {
crm_trace("Unchanged %s[%s] from %s is %s", attr, host, peer->uname, value);
}
a->changed |= changed;
if(changed) {
if(a->timer) {
crm_trace("Delayed write out (%dms) for %s", a->timeout_ms, a->id);
mainloop_timer_start(a->timer);
} else {
write_or_elect_attribute(a);
}
}
/* this only involves cluster nodes. */
if(v->nodeid == 0 && (v->is_remote == FALSE)) {
if(crm_element_value_int(xml, F_ATTRD_HOST_ID, (int*)&v->nodeid) == 0) {
/* Create the name/id association */
crm_node_t *peer = crm_get_peer(v->nodeid, host);
crm_trace("We know %s's node id now: %s", peer->uname, peer->uuid);
if(election_state(writer) == election_won) {
write_attributes(FALSE, TRUE);
return;
}
}
}
}
void
write_or_elect_attribute(attribute_t *a)
{
enum election_result rc = election_state(writer);
if(rc == election_won) {
write_attribute(a);
} else if(rc == election_in_progress) {
crm_trace("Election in progress to determine who will write out %s", a->id);
} else if(peer_writer == NULL) {
crm_info("Starting an election to determine who will write out %s", a->id);
election_vote(writer);
} else {
crm_trace("%s will write out %s, we are in state %d", peer_writer, a->id, rc);
}
}
gboolean
attrd_election_cb(gpointer user_data)
{
crm_trace("Election complete");
free(peer_writer);
peer_writer = strdup(attrd_cluster->uname);
/* Update the peers after an election */
attrd_peer_sync(NULL, NULL);
/* Update the CIB after an election */
write_attributes(TRUE, FALSE);
return FALSE;
}
void
attrd_peer_change_cb(enum crm_status_type kind, crm_node_t *peer, const void *data)
{
if ((kind == crm_status_nstate) || (kind == crm_status_rstate)) {
if (safe_str_eq(peer->state, CRM_NODE_MEMBER)) {
if ((election_state(writer) == election_won)) {
attrd_peer_sync(peer, NULL);
}
} else {
attrd_peer_remove(peer->id, peer->uname, FALSE, __FUNCTION__);
if (peer_writer && safe_str_eq(peer->uname, peer_writer)) {
free(peer_writer);
peer_writer = NULL;
crm_notice("Lost attribute writer %s", peer->uname);
}
}
- } else if (kind == crm_status_processes) {
- crm_update_peer_state(__FUNCTION__, peer,
- is_set(peer->processes, crm_proc_cpg)?
- CRM_NODE_MEMBER : CRM_NODE_LOST, 0);
}
}
static void
attrd_cib_callback(xmlNode * msg, int call_id, int rc, xmlNode * output, void *user_data)
{
int level = LOG_ERR;
GHashTableIter iter;
const char *peer = NULL;
attribute_value_t *v = NULL;
char *name = user_data;
attribute_t *a = g_hash_table_lookup(attributes, name);
if(a == NULL) {
crm_info("Attribute %s no longer exists", name);
goto done;
}
a->update = 0;
if (rc == pcmk_ok && call_id < 0) {
rc = call_id;
}
switch (rc) {
case pcmk_ok:
level = LOG_INFO;
last_cib_op_done = call_id;
break;
case -pcmk_err_diff_failed: /* When an attr changes while the CIB is syncing */
case -ETIME: /* When an attr changes while there is a DC election */
case -ENXIO: /* When an attr changes while the CIB is syncing a
* newer config from a node that just came up
*/
level = LOG_WARNING;
break;
}
do_crm_log(level, "Update %d for %s: %s (%d)", call_id, name, pcmk_strerror(rc), rc);
g_hash_table_iter_init(&iter, a->values);
while (g_hash_table_iter_next(&iter, (gpointer *) & peer, (gpointer *) & v)) {
do_crm_log(level, "Update %d for %s[%s]=%s: %s (%d)", call_id, a->id, peer, v->requested, pcmk_strerror(rc), rc);
if(rc == pcmk_ok) {
free(v->stored);
v->stored = v->requested;
v->requested = NULL;
} else {
free(v->requested);
v->requested = NULL;
a->changed = TRUE; /* Attempt write out again */
}
}
done:
free(name);
if(a && a->changed && election_state(writer) == election_won) {
write_attribute(a);
}
}
void
write_attributes(bool all, bool peer_discovered)
{
GHashTableIter iter;
attribute_t *a = NULL;
g_hash_table_iter_init(&iter, attributes);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) & a)) {
if (peer_discovered && a->unknown_peer_uuids) {
/* a new peer uuid has been discovered, try writing this attribute again. */
a->changed = TRUE;
}
if(all || a->changed) {
write_attribute(a);
} else {
crm_debug("Skipping unchanged attribute %s", a->id);
}
}
}
static void
build_update_element(xmlNode *parent, attribute_t *a, const char *nodeid, const char *value)
{
char *set = NULL;
char *uuid = NULL;
xmlNode *xml_obj = NULL;
if(a->set) {
set = strdup(a->set);
} else {
set = crm_strdup_printf("%s-%s", XML_CIB_TAG_STATUS, nodeid);
}
if(a->uuid) {
uuid = strdup(a->uuid);
} else {
int lpc;
uuid = crm_strdup_printf("%s-%s", set, a->id);
/* Minimal attempt at sanitizing automatic IDs */
for (lpc = 0; uuid[lpc] != 0; lpc++) {
switch (uuid[lpc]) {
case ':':
uuid[lpc] = '.';
}
}
}
xml_obj = create_xml_node(parent, XML_CIB_TAG_STATE);
crm_xml_add(xml_obj, XML_ATTR_ID, nodeid);
xml_obj = create_xml_node(xml_obj, XML_TAG_TRANSIENT_NODEATTRS);
crm_xml_add(xml_obj, XML_ATTR_ID, nodeid);
xml_obj = create_xml_node(xml_obj, XML_TAG_ATTR_SETS);
crm_xml_add(xml_obj, XML_ATTR_ID, set);
xml_obj = create_xml_node(xml_obj, XML_CIB_TAG_NVPAIR);
crm_xml_add(xml_obj, XML_ATTR_ID, uuid);
crm_xml_add(xml_obj, XML_NVPAIR_ATTR_NAME, a->id);
if(value) {
crm_xml_add(xml_obj, XML_NVPAIR_ATTR_VALUE, value);
} else {
crm_xml_add(xml_obj, XML_NVPAIR_ATTR_VALUE, "");
crm_xml_add(xml_obj, "__delete__", XML_NVPAIR_ATTR_VALUE);
}
free(uuid);
free(set);
}
void
write_attribute(attribute_t *a)
{
int private_updates = 0, cib_updates = 0;
xmlNode *xml_top = NULL;
attribute_value_t *v = NULL;
GHashTableIter iter;
enum cib_call_options flags = cib_quorum_override;
if (a == NULL) {
return;
}
/* If this attribute will be written to the CIB ... */
if (!a->is_private) {
/* Defer the write if now's not a good time */
if (the_cib == NULL) {
crm_info("Write out of '%s' delayed: cib not connected", a->id);
return;
} else if (a->update && (a->update < last_cib_op_done)) {
crm_info("Write out of '%s' continuing: update %d considered lost", a->id, a->update);
} else if (a->update) {
crm_info("Write out of '%s' delayed: update %d in progress", a->id, a->update);
return;
} else if (mainloop_timer_running(a->timer)) {
crm_info("Write out of '%s' delayed: timer is running", a->id);
return;
}
/* Initialize the status update XML */
xml_top = create_xml_node(NULL, XML_CIB_TAG_STATUS);
}
/* Attribute will be written shortly, so clear changed flag */
a->changed = FALSE;
/* We will check all peers' uuids shortly, so initialize this to false */
a->unknown_peer_uuids = FALSE;
/* Iterate over each peer value of this attribute */
g_hash_table_iter_init(&iter, a->values);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) & v)) {
crm_node_t *peer = crm_get_peer_full(v->nodeid, v->nodename, CRM_GET_PEER_REMOTE|CRM_GET_PEER_CLUSTER);
/* If the value's peer info does not correspond to a peer, ignore it */
if (peer == NULL) {
crm_notice("Update error (peer not found): %s[%s]=%s failed (host=%p)",
v->nodename, a->id, v->current, peer);
continue;
}
/* If we're just learning the peer's node id, remember it */
if (peer->id && (v->nodeid == 0)) {
crm_trace("Updating value's nodeid");
v->nodeid = peer->id;
}
/* If this is a private attribute, no update needs to be sent */
if (a->is_private) {
private_updates++;
continue;
}
/* If the peer is found, but its uuid is unknown, defer write */
if (peer->uuid == NULL) {
a->unknown_peer_uuids = TRUE;
crm_notice("Update error (unknown peer uuid, retry will be attempted once uuid is discovered): %s[%s]=%s failed (host=%p)",
v->nodename, a->id, v->current, peer);
continue;
}
/* Add this value to status update XML */
crm_debug("Update: %s[%s]=%s (%s %u %u %s)", v->nodename, a->id,
v->current, peer->uuid, peer->id, v->nodeid, peer->uname);
build_update_element(xml_top, a, peer->uuid, v->current);
cib_updates++;
free(v->requested);
v->requested = NULL;
if (v->current) {
v->requested = strdup(v->current);
} else {
/* Older attrd versions don't know about the cib_mixed_update
* flag so make sure it goes to the local cib which does
*/
flags |= cib_mixed_update|cib_scope_local;
}
}
if (private_updates) {
crm_info("Processed %d private change%s for %s, id=%s, set=%s",
private_updates, ((private_updates == 1)? "" : "s"),
a->id, (a->uuid? a->uuid : "<n/a>"), a->set);
}
if (cib_updates) {
crm_log_xml_trace(xml_top, __FUNCTION__);
a->update = cib_internal_op(the_cib, CIB_OP_MODIFY, NULL, XML_CIB_TAG_STATUS, xml_top, NULL,
flags, a->user);
crm_info("Sent update %d with %d changes for %s, id=%s, set=%s",
a->update, cib_updates, a->id, (a->uuid? a->uuid : "<n/a>"), a->set);
the_cib->cmds->register_callback(
the_cib, a->update, 120, FALSE, strdup(a->id), "attrd_cib_callback", attrd_cib_callback);
}
free_xml(xml_top);
}
diff --git a/cib/main.c b/cib/main.c
index 3261415fda..2a480545d8 100644
--- a/cib/main.c
+++ b/cib/main.c
@@ -1,585 +1,576 @@
/*
* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
*
* 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 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <sys/param.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/utsname.h>
#include <stdlib.h>
#include <errno.h>
#include <fcntl.h>
#include <crm/crm.h>
#include <crm/cib/internal.h>
#include <crm/msg_xml.h>
#include <crm/cluster/internal.h>
#include <crm/common/xml.h>
#include <crm/common/mainloop.h>
#include <cibio.h>
#include <callbacks.h>
#include <pwd.h>
#include <grp.h>
#include "common.h"
#if HAVE_LIBXML2
# include <libxml/parser.h>
#endif
#ifdef HAVE_GETOPT_H
# include <getopt.h>
#endif
#if HAVE_BZLIB_H
# include <bzlib.h>
#endif
extern int init_remote_listener(int port, gboolean encrypted);
gboolean cib_shutdown_flag = FALSE;
int cib_status = pcmk_ok;
crm_cluster_t crm_cluster;
#if SUPPORT_HEARTBEAT
oc_ev_t *cib_ev_token;
ll_cluster_t *hb_conn = NULL;
extern void oc_ev_special(const oc_ev_t *, oc_ev_class_t, int);
gboolean cib_register_ha(ll_cluster_t * hb_cluster, const char *client_name);
#else
void *hb_conn = NULL;
#endif
extern void terminate_cib(const char *caller, gboolean fast);
GMainLoop *mainloop = NULL;
const char *cib_root = NULL;
char *cib_our_uname = NULL;
gboolean preserve_status = FALSE;
gboolean cib_writes_enabled = TRUE;
int remote_fd = 0;
int remote_tls_fd = 0;
int cib_init(void);
void cib_shutdown(int nsig);
gboolean startCib(const char *filename);
extern int write_cib_contents(gpointer p);
GHashTable *config_hash = NULL;
GHashTable *local_notify_queue = NULL;
char *channel1 = NULL;
char *channel2 = NULL;
char *channel3 = NULL;
char *channel4 = NULL;
char *channel5 = NULL;
#define OPTARGS "maswr:V?"
void cib_cleanup(void);
static void
cib_enable_writes(int nsig)
{
crm_info("(Re)enabling disk writes");
cib_writes_enabled = TRUE;
}
static void
log_cib_client(gpointer key, gpointer value, gpointer user_data)
{
crm_info("Client %s", crm_client_name(value));
}
/* *INDENT-OFF* */
static struct crm_option long_options[] = {
/* Top-level Options */
{"help", 0, 0, '?', "\tThis text"},
{"verbose", 0, 0, 'V', "\tIncrease debug output"},
{"per-action-cib", 0, 0, 'a', "\tAdvanced use only"},
{"stand-alone", 0, 0, 's', "\tAdvanced use only"},
{"disk-writes", 0, 0, 'w', "\tAdvanced use only"},
{"cib-root", 1, 0, 'r', "\tAdvanced use only"},
{0, 0, 0, 0}
};
/* *INDENT-ON* */
int
main(int argc, char **argv)
{
int flag;
int rc = 0;
int index = 0;
int argerr = 0;
struct passwd *pwentry = NULL;
crm_log_preinit(NULL, argc, argv);
crm_set_options(NULL, "[options]",
long_options, "Daemon for storing and replicating the cluster configuration");
crm_peer_init();
mainloop_add_signal(SIGTERM, cib_shutdown);
mainloop_add_signal(SIGPIPE, cib_enable_writes);
cib_writer = mainloop_add_trigger(G_PRIORITY_LOW, write_cib_contents, NULL);
while (1) {
flag = crm_get_option(argc, argv, &index);
if (flag == -1)
break;
switch (flag) {
case 'V':
crm_bump_log_level(argc, argv);
break;
case 's':
stand_alone = TRUE;
preserve_status = TRUE;
cib_writes_enabled = FALSE;
pwentry = getpwnam(CRM_DAEMON_USER);
CRM_CHECK(pwentry != NULL,
crm_perror(LOG_ERR, "Invalid uid (%s) specified", CRM_DAEMON_USER);
return 100);
rc = setgid(pwentry->pw_gid);
if (rc < 0) {
crm_perror(LOG_ERR, "Could not set group to %d", pwentry->pw_gid);
return 100;
}
rc = initgroups(CRM_DAEMON_GROUP, pwentry->pw_gid);
if (rc < 0) {
crm_perror(LOG_ERR, "Could not setup groups for user %d", pwentry->pw_uid);
return 100;
}
rc = setuid(pwentry->pw_uid);
if (rc < 0) {
crm_perror(LOG_ERR, "Could not set user to %d", pwentry->pw_uid);
return 100;
}
break;
case '?': /* Help message */
crm_help(flag, EX_OK);
break;
case 'w':
cib_writes_enabled = TRUE;
break;
case 'r':
cib_root = optarg;
break;
case 'm':
cib_metadata();
return 0;
default:
++argerr;
break;
}
}
if (argc - optind == 1 && safe_str_eq("metadata", argv[optind])) {
cib_metadata();
return 0;
}
if (optind > argc) {
++argerr;
}
if (argerr) {
crm_help('?', EX_USAGE);
}
crm_log_init(NULL, LOG_INFO, TRUE, FALSE, argc, argv, FALSE);
if (cib_root == NULL) {
char *path = crm_strdup_printf("%s/cib.xml", CRM_CONFIG_DIR);
char *legacy = crm_strdup_printf("%s/cib.xml", CRM_LEGACY_CONFIG_DIR);
if (g_file_test(path, G_FILE_TEST_EXISTS)) {
cib_root = CRM_CONFIG_DIR;
} else if (g_file_test(legacy, G_FILE_TEST_EXISTS)) {
cib_root = CRM_LEGACY_CONFIG_DIR;
crm_notice("Using legacy config location: %s", cib_root);
} else {
cib_root = CRM_CONFIG_DIR;
crm_notice("Using new config location: %s", cib_root);
}
free(legacy);
free(path);
} else {
crm_notice("Using custom config location: %s", cib_root);
}
if (crm_is_writable(cib_root, NULL, CRM_DAEMON_USER, CRM_DAEMON_GROUP, FALSE) == FALSE) {
crm_err("Bad permissions on %s. Terminating", cib_root);
fprintf(stderr, "ERROR: Bad permissions on %s. See logs for details\n", cib_root);
fflush(stderr);
return 100;
}
/* read local config file */
rc = cib_init();
CRM_CHECK(crm_hash_table_size(client_connections) == 0,
crm_warn("Not all clients gone at exit"));
g_hash_table_foreach(client_connections, log_cib_client, NULL);
cib_cleanup();
#if SUPPORT_HEARTBEAT
if (hb_conn) {
hb_conn->llc_ops->delete(hb_conn);
}
#endif
crm_info("Done");
return rc;
}
void
cib_cleanup(void)
{
crm_peer_destroy();
if (local_notify_queue) {
g_hash_table_destroy(local_notify_queue);
}
crm_client_cleanup();
g_hash_table_destroy(config_hash);
free(cib_our_uname);
free(channel1);
free(channel2);
free(channel3);
free(channel4);
free(channel5);
}
unsigned long cib_num_ops = 0;
const char *cib_stat_interval = "10min";
unsigned long cib_num_local = 0, cib_num_updates = 0, cib_num_fail = 0;
unsigned long cib_bad_connects = 0, cib_num_timeouts = 0;
#if SUPPORT_HEARTBEAT
gboolean ccm_connect(void);
static void
ccm_connection_destroy(gpointer user_data)
{
crm_err("CCM connection failed... blocking while we reconnect");
CRM_ASSERT(ccm_connect());
return;
}
static void *ccm_library = NULL;
gboolean
ccm_connect(void)
{
gboolean did_fail = TRUE;
int num_ccm_fails = 0;
int max_ccm_fails = 30;
int ret;
int cib_ev_fd;
int (*ccm_api_register) (oc_ev_t ** token) =
find_library_function(&ccm_library, CCM_LIBRARY, "oc_ev_register", 1);
int (*ccm_api_set_callback) (const oc_ev_t * token,
oc_ev_class_t class,
oc_ev_callback_t * fn,
oc_ev_callback_t ** prev_fn) =
find_library_function(&ccm_library, CCM_LIBRARY, "oc_ev_set_callback", 1);
void (*ccm_api_special) (const oc_ev_t *, oc_ev_class_t, int) =
find_library_function(&ccm_library, CCM_LIBRARY, "oc_ev_special", 1);
int (*ccm_api_activate) (const oc_ev_t * token, int *fd) =
find_library_function(&ccm_library, CCM_LIBRARY, "oc_ev_activate", 1);
int (*ccm_api_unregister) (oc_ev_t * token) =
find_library_function(&ccm_library, CCM_LIBRARY, "oc_ev_unregister", 1);
static struct mainloop_fd_callbacks ccm_fd_callbacks = {
.dispatch = cib_ccm_dispatch,
.destroy = ccm_connection_destroy,
};
while (did_fail) {
did_fail = FALSE;
crm_info("Registering with CCM...");
ret = (*ccm_api_register) (&cib_ev_token);
if (ret != 0) {
did_fail = TRUE;
}
if (did_fail == FALSE) {
crm_trace("Setting up CCM callbacks");
ret = (*ccm_api_set_callback) (cib_ev_token, OC_EV_MEMB_CLASS,
cib_ccm_msg_callback, NULL);
if (ret != 0) {
crm_warn("CCM callback not set");
did_fail = TRUE;
}
}
if (did_fail == FALSE) {
(*ccm_api_special) (cib_ev_token, OC_EV_MEMB_CLASS, 0);
crm_trace("Activating CCM token");
ret = (*ccm_api_activate) (cib_ev_token, &cib_ev_fd);
if (ret != 0) {
crm_warn("CCM Activation failed");
did_fail = TRUE;
}
}
if (did_fail) {
num_ccm_fails++;
(*ccm_api_unregister) (cib_ev_token);
if (num_ccm_fails < max_ccm_fails) {
crm_warn("CCM Connection failed %d times (%d max)", num_ccm_fails, max_ccm_fails);
sleep(3);
} else {
crm_err("CCM Activation failed %d (max) times", num_ccm_fails);
return FALSE;
}
}
}
crm_debug("CCM Activation passed... all set to go!");
mainloop_add_fd("heartbeat-ccm", G_PRIORITY_MEDIUM, cib_ev_fd, cib_ev_token, &ccm_fd_callbacks);
return TRUE;
}
#endif
#if SUPPORT_COROSYNC
static void
cib_cs_dispatch(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;
xmlNode *xml = NULL;
const char *from = NULL;
char *data = pcmk_message_common_cs(handle, nodeid, pid, msg, &kind, &from);
if(data == NULL) {
return;
}
if (kind == crm_class_cluster) {
xml = string2xml(data);
if (xml == NULL) {
crm_err("Invalid XML: '%.120s'", data);
free(data);
return;
}
crm_xml_add(xml, F_ORIG, from);
/* crm_xml_add_int(xml, F_SEQ, wrapper->id); */
cib_peer_callback(xml, NULL);
}
free_xml(xml);
free(data);
}
static void
cib_cs_destroy(gpointer user_data)
{
if (cib_shutdown_flag) {
crm_info("Corosync disconnection complete");
} else {
crm_err("Corosync connection lost! Exiting.");
terminate_cib(__FUNCTION__, TRUE);
}
}
#endif
static void
cib_peer_update_callback(enum crm_status_type type, crm_node_t * node, const void *data)
{
- if (type == crm_status_processes) {
- crm_update_peer_state(__FUNCTION__, node, is_set(node->processes, crm_proc_cpg)?CRM_NODE_MEMBER:CRM_NODE_LOST, 0);
- }
-
if ((type == crm_status_processes) && legacy_mode
&& is_not_set(node->processes, crm_get_cluster_proc())) {
uint32_t old = 0;
if (data) {
old = *(const uint32_t *)data;
}
if ((node->processes ^ old) & crm_proc_cpg) {
crm_info("Attempting to disable legacy mode after %s left the cluster", node->uname);
legacy_mode = FALSE;
}
}
if (cib_shutdown_flag && crm_active_peers() < 2 && crm_hash_table_size(client_connections) == 0) {
crm_info("No more peers");
terminate_cib(__FUNCTION__, FALSE);
}
-
- if(type == crm_status_nstate && node->id && safe_str_eq(node->state, CRM_NODE_LOST)) {
- /* Avoid conflicts, keep the membership list to active members */
- reap_crm_member(node->id, NULL);
- }
}
#if SUPPORT_HEARTBEAT
static void
cib_ha_connection_destroy(gpointer user_data)
{
if (cib_shutdown_flag) {
crm_info("Heartbeat disconnection complete... exiting");
terminate_cib(__FUNCTION__, FALSE);
} else {
crm_err("Heartbeat connection lost! Exiting.");
terminate_cib(__FUNCTION__, TRUE);
}
}
#endif
int
cib_init(void)
{
if (is_openais_cluster()) {
#if SUPPORT_COROSYNC
crm_cluster.destroy = cib_cs_destroy;
crm_cluster.cpg.cpg_deliver_fn = cib_cs_dispatch;
crm_cluster.cpg.cpg_confchg_fn = pcmk_cpg_membership;
#endif
} else if (is_heartbeat_cluster()) {
#if SUPPORT_HEARTBEAT
crm_cluster.hb_dispatch = cib_ha_peer_callback;
crm_cluster.destroy = cib_ha_connection_destroy;
#endif
}
config_hash =
g_hash_table_new_full(crm_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str);
if (startCib("cib.xml") == FALSE) {
crm_crit("Cannot start CIB... terminating");
crm_exit(ENODATA);
}
if (stand_alone == FALSE) {
if (crm_cluster_connect(&crm_cluster) == FALSE) {
crm_crit("Cannot sign in to the cluster... terminating");
crm_exit(DAEMON_RESPAWN_STOP);
}
cib_our_uname = crm_cluster.uname;
if (is_openais_cluster()) {
crm_set_status_callback(&cib_peer_update_callback);
}
#if SUPPORT_HEARTBEAT
if (is_heartbeat_cluster()) {
gboolean was_error = FALSE;
hb_conn = crm_cluster.hb_conn;
if (was_error == FALSE) {
if (HA_OK !=
hb_conn->llc_ops->set_cstatus_callback(hb_conn, cib_client_status_callback,
hb_conn)) {
crm_err("Cannot set cstatus callback: %s", hb_conn->llc_ops->errmsg(hb_conn));
was_error = TRUE;
}
}
if (was_error == FALSE) {
was_error = (ccm_connect() == FALSE);
}
if (was_error == FALSE) {
/* Async get client status information in the cluster */
crm_info("Requesting the list of configured nodes");
hb_conn->llc_ops->client_status(hb_conn, NULL, CRM_SYSTEM_CIB, -1);
}
}
#endif
} else {
cib_our_uname = strdup("localhost");
}
cib_ipc_servers_init(&ipcs_ro,
&ipcs_rw,
&ipcs_shm,
&ipc_ro_callbacks,
&ipc_rw_callbacks);
if (stand_alone) {
cib_is_master = TRUE;
}
/* Create the mainloop and run it... */
mainloop = g_main_new(FALSE);
crm_info("Starting %s mainloop", crm_system_name);
g_main_run(mainloop);
cib_ipc_servers_destroy(ipcs_ro, ipcs_rw, ipcs_shm);
return crm_exit(pcmk_ok);
}
gboolean
startCib(const char *filename)
{
gboolean active = FALSE;
xmlNode *cib = readCibXmlFile(cib_root, filename, !preserve_status);
CRM_ASSERT(cib != NULL);
if (activateCibXml(cib, TRUE, "start") == 0) {
int port = 0;
const char *port_s = NULL;
active = TRUE;
cib_read_config(config_hash, cib);
port_s = crm_element_value(cib, "remote-tls-port");
if (port_s) {
port = crm_parse_int(port_s, "0");
remote_tls_fd = init_remote_listener(port, TRUE);
}
port_s = crm_element_value(cib, "remote-clear-port");
if (port_s) {
port = crm_parse_int(port_s, "0");
remote_fd = init_remote_listener(port, FALSE);
}
crm_info("CIB Initialization completed successfully");
}
return active;
}
diff --git a/crmd/control.c b/crmd/control.c
index 2fd8a07526..f066d4db6d 100644
--- a/crmd/control.c
+++ b/crmd/control.c
@@ -1,1104 +1,1105 @@
/*
* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
*
* 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 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <sys/param.h>
#include <crm/crm.h>
#include <crm/msg_xml.h>
#include <crm/pengine/rules.h>
#include <crm/cluster/internal.h>
#include <crm/cluster/election.h>
#include <crm/common/ipcs.h>
#include <crmd.h>
#include <crmd_fsa.h>
#include <fsa_proto.h>
#include <crmd_messages.h>
#include <crmd_callbacks.h>
#include <crmd_lrm.h>
#include <tengine.h>
#include <throttle.h>
#include <sys/types.h>
#include <sys/stat.h>
qb_ipcs_service_t *ipcs = NULL;
extern gboolean crm_connect_corosync(crm_cluster_t * cluster);
extern void crmd_ha_connection_destroy(gpointer user_data);
void crm_shutdown(int nsig);
gboolean crm_read_options(gpointer user_data);
gboolean fsa_has_quorum = FALSE;
crm_trigger_t *fsa_source = NULL;
crm_trigger_t *config_read = NULL;
bool no_quorum_suicide_escalation = FALSE;
static gboolean
election_timeout_popped(gpointer data)
{
/* Not everyone voted */
crm_info("Election failed: Declaring ourselves the winner");
register_fsa_input(C_TIMER_POPPED, I_ELECTION_DC, NULL);
return FALSE;
}
/* A_HA_CONNECT */
void
do_ha_control(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state,
enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
gboolean registered = FALSE;
static crm_cluster_t *cluster = NULL;
if (cluster == NULL) {
cluster = calloc(1, sizeof(crm_cluster_t));
}
if (action & A_HA_DISCONNECT) {
crm_cluster_disconnect(cluster);
crm_info("Disconnected from the cluster");
set_bit(fsa_input_register, R_HA_DISCONNECTED);
}
if (action & A_HA_CONNECT) {
crm_set_status_callback(&peer_update_callback);
+ crm_set_autoreap(FALSE);
if (is_openais_cluster()) {
#if SUPPORT_COROSYNC
registered = crm_connect_corosync(cluster);
#endif
} else if (is_heartbeat_cluster()) {
#if SUPPORT_HEARTBEAT
cluster->destroy = crmd_ha_connection_destroy;
cluster->hb_dispatch = crmd_ha_msg_callback;
registered = crm_cluster_connect(cluster);
fsa_cluster_conn = cluster->hb_conn;
crm_trace("Be informed of Node Status changes");
if (registered &&
fsa_cluster_conn->llc_ops->set_nstatus_callback(fsa_cluster_conn,
crmd_ha_status_callback,
fsa_cluster_conn) != HA_OK) {
crm_err("Cannot set nstatus callback: %s",
fsa_cluster_conn->llc_ops->errmsg(fsa_cluster_conn));
registered = FALSE;
}
crm_trace("Be informed of CRM Client Status changes");
if (registered &&
fsa_cluster_conn->llc_ops->set_cstatus_callback(fsa_cluster_conn,
crmd_client_status_callback,
fsa_cluster_conn) != HA_OK) {
crm_err("Cannot set cstatus callback: %s",
fsa_cluster_conn->llc_ops->errmsg(fsa_cluster_conn));
registered = FALSE;
}
if (registered) {
crm_trace("Requesting an initial dump of CRMD client_status");
fsa_cluster_conn->llc_ops->client_status(fsa_cluster_conn, NULL, CRM_SYSTEM_CRMD,
-1);
}
#endif
}
fsa_election = election_init(NULL, cluster->uname, 60000/*60s*/, election_timeout_popped);
fsa_our_uname = cluster->uname;
fsa_our_uuid = cluster->uuid;
if(cluster->uuid == NULL) {
crm_err("Could not obtain local uuid");
registered = FALSE;
}
if (registered == FALSE) {
set_bit(fsa_input_register, R_HA_DISCONNECTED);
register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
return;
}
populate_cib_nodes(node_update_none, __FUNCTION__);
clear_bit(fsa_input_register, R_HA_DISCONNECTED);
crm_info("Connected to the cluster");
}
if (action & ~(A_HA_CONNECT | A_HA_DISCONNECT)) {
crm_err("Unexpected action %s in %s", fsa_action2string(action), __FUNCTION__);
}
}
static bool
need_spawn_pengine_from_crmd(void)
{
static int result = -1;
if (result != -1)
return result;
if (!is_heartbeat_cluster()) {
result = 0;
return result;
}
/* NULL, or "strange" value: rather spawn from here. */
result = TRUE;
crm_str_to_boolean(daemon_option("crmd_spawns_pengine"), &result);
return result;
}
/* A_SHUTDOWN */
void
do_shutdown(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
/* just in case */
set_bit(fsa_input_register, R_SHUTDOWN);
if (need_spawn_pengine_from_crmd()) {
if (is_set(fsa_input_register, pe_subsystem->flag_connected)) {
crm_info("Terminating the %s", pe_subsystem->name);
if (stop_subsystem(pe_subsystem, TRUE) == FALSE) {
/* its gone... */
crm_err("Faking %s exit", pe_subsystem->name);
clear_bit(fsa_input_register, pe_subsystem->flag_connected);
} else {
crm_info("Waiting for subsystems to exit");
crmd_fsa_stall(FALSE);
}
}
crm_info("All subsystems stopped, continuing");
}
if (stonith_api) {
/* Prevent it from comming up again */
clear_bit(fsa_input_register, R_ST_REQUIRED);
crm_info("Disconnecting STONITH...");
stonith_api->cmds->disconnect(stonith_api);
}
}
/* A_SHUTDOWN_REQ */
void
do_shutdown_req(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state,
enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
xmlNode *msg = NULL;
crm_info("Sending shutdown request to %s", crm_str(fsa_our_dc));
msg = create_request(CRM_OP_SHUTDOWN_REQ, NULL, NULL, CRM_SYSTEM_CRMD, CRM_SYSTEM_CRMD, NULL);
/* set_bit(fsa_input_register, R_STAYDOWN); */
if (send_cluster_message(NULL, crm_msg_crmd, msg, TRUE) == FALSE) {
register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
}
free_xml(msg);
}
extern crm_ipc_t *attrd_ipc;
extern char *max_generation_from;
extern xmlNode *max_generation_xml;
extern GHashTable *resource_history;
extern GHashTable *voted;
extern GHashTable *reload_hash;
extern char *te_client_id;
void log_connected_client(gpointer key, gpointer value, gpointer user_data);
void
log_connected_client(gpointer key, gpointer value, gpointer user_data)
{
crm_client_t *client = value;
crm_err("%s is still connected at exit", crm_client_name(client));
}
int
crmd_fast_exit(int rc)
{
if (is_set(fsa_input_register, R_STAYDOWN)) {
crm_warn("Inhibiting respawn: %d -> %d", rc, 100);
rc = 100;
}
if (rc == pcmk_ok && is_set(fsa_input_register, R_IN_RECOVERY)) {
crm_err("Could not recover from internal error");
rc = pcmk_err_generic;
}
return crm_exit(rc);
}
int
crmd_exit(int rc)
{
GListPtr gIter = NULL;
GMainLoop *mloop = crmd_mainloop;
static bool in_progress = FALSE;
if(in_progress && rc == 0) {
crm_debug("Exit is already in progress");
return rc;
} else if(in_progress) {
crm_notice("Error during shutdown process, terminating now: %s (%d)", pcmk_strerror(rc), rc);
crm_write_blackbox(SIGTRAP, NULL);
crmd_fast_exit(rc);
}
in_progress = TRUE;
crm_trace("Preparing to exit: %d", rc);
/* Suppress secondary errors resulting from us disconnecting everything */
set_bit(fsa_input_register, R_HA_DISCONNECTED);
/* Close all IPC servers and clients to ensure any and all shared memory files are cleaned up */
if(ipcs) {
crm_trace("Closing IPC server");
mainloop_del_ipc_server(ipcs);
ipcs = NULL;
}
if (attrd_ipc) {
crm_trace("Closing attrd connection");
crm_ipc_close(attrd_ipc);
crm_ipc_destroy(attrd_ipc);
attrd_ipc = NULL;
}
if (pe_subsystem && pe_subsystem->client && pe_subsystem->client->ipcs) {
crm_trace("Disconnecting Policy Engine");
qb_ipcs_disconnect(pe_subsystem->client->ipcs);
}
if(stonith_api) {
crm_trace("Disconnecting fencing API");
clear_bit(fsa_input_register, R_ST_REQUIRED);
stonith_api->cmds->free(stonith_api); stonith_api = NULL;
}
if (rc == pcmk_ok && crmd_mainloop == NULL) {
crm_debug("No mainloop detected");
rc = EPROTO;
}
/* On an error, just get out.
*
* Otherwise, make the effort to have mainloop exit gracefully so
* that it (mostly) cleans up after itself and valgrind has less
* to report on - allowing real errors stand out
*/
if(rc != pcmk_ok) {
crm_notice("Forcing immediate exit: %s (%d)", pcmk_strerror(rc), rc);
crm_write_blackbox(SIGTRAP, NULL);
return crmd_fast_exit(rc);
}
/* Clean up as much memory as possible for valgrind */
for (gIter = fsa_message_queue; gIter != NULL; gIter = gIter->next) {
fsa_data_t *fsa_data = gIter->data;
crm_info("Dropping %s: [ state=%s cause=%s origin=%s ]",
fsa_input2string(fsa_data->fsa_input),
fsa_state2string(fsa_state),
fsa_cause2string(fsa_data->fsa_cause), fsa_data->origin);
delete_fsa_input(fsa_data);
}
clear_bit(fsa_input_register, R_MEMBERSHIP);
g_list_free(fsa_message_queue); fsa_message_queue = NULL;
free(pe_subsystem); pe_subsystem = NULL;
free(te_subsystem); te_subsystem = NULL;
free(cib_subsystem); cib_subsystem = NULL;
if (reload_hash) {
crm_trace("Destroying reload cache with %d members", g_hash_table_size(reload_hash));
g_hash_table_destroy(reload_hash); reload_hash = NULL;
}
election_fini(fsa_election);
fsa_election = NULL;
cib_delete(fsa_cib_conn);
fsa_cib_conn = NULL;
verify_stopped(fsa_state, LOG_WARNING);
clear_bit(fsa_input_register, R_LRM_CONNECTED);
lrm_state_destroy_all();
/* This basically will not work, since mainloop has a reference to it */
mainloop_destroy_trigger(fsa_source); fsa_source = NULL;
mainloop_destroy_trigger(config_read); config_read = NULL;
mainloop_destroy_trigger(stonith_reconnect); stonith_reconnect = NULL;
mainloop_destroy_trigger(transition_trigger); transition_trigger = NULL;
crm_client_cleanup();
crm_peer_destroy();
crm_timer_stop(transition_timer);
crm_timer_stop(integration_timer);
crm_timer_stop(finalization_timer);
crm_timer_stop(election_trigger);
election_timeout_stop(fsa_election);
crm_timer_stop(shutdown_escalation_timer);
crm_timer_stop(wait_timer);
crm_timer_stop(recheck_timer);
free(transition_timer); transition_timer = NULL;
free(integration_timer); integration_timer = NULL;
free(finalization_timer); finalization_timer = NULL;
free(election_trigger); election_trigger = NULL;
election_fini(fsa_election);
free(shutdown_escalation_timer); shutdown_escalation_timer = NULL;
free(wait_timer); wait_timer = NULL;
free(recheck_timer); recheck_timer = NULL;
free(fsa_our_dc_version); fsa_our_dc_version = NULL;
free(fsa_our_uname); fsa_our_uname = NULL;
free(fsa_our_uuid); fsa_our_uuid = NULL;
free(fsa_our_dc); fsa_our_dc = NULL;
free(fsa_cluster_name); fsa_cluster_name = NULL;
free(te_uuid); te_uuid = NULL;
free(te_client_id); te_client_id = NULL;
free(fsa_pe_ref); fsa_pe_ref = NULL;
free(failed_stop_offset); failed_stop_offset = NULL;
free(failed_start_offset); failed_start_offset = NULL;
free(max_generation_from); max_generation_from = NULL;
free_xml(max_generation_xml); max_generation_xml = NULL;
mainloop_destroy_signal(SIGPIPE);
mainloop_destroy_signal(SIGUSR1);
mainloop_destroy_signal(SIGTERM);
mainloop_destroy_signal(SIGTRAP);
mainloop_destroy_signal(SIGCHLD);
if (mloop) {
int lpc = 0;
GMainContext *ctx = g_main_loop_get_context(crmd_mainloop);
/* Don't re-enter this block */
crmd_mainloop = NULL;
crm_trace("Draining mainloop %d %d", g_main_loop_is_running(mloop), g_main_context_pending(ctx));
while(g_main_context_pending(ctx) && lpc < 10) {
lpc++;
crm_trace("Iteration %d", lpc);
g_main_context_dispatch(ctx);
}
crm_trace("Closing mainloop %d %d", g_main_loop_is_running(mloop), g_main_context_pending(ctx));
g_main_loop_quit(mloop);
#if SUPPORT_HEARTBEAT
/* Do this only after g_main_loop_quit().
*
* This interface was broken (incomplete) since it was introduced.
* ->delete() does cleanup and free most of it, but it does not
* actually remove and destroy the corresponding GSource, so the next
* prepare/check iteratioin would find a corrupt (because partially
* freed) GSource, and segfault.
*
* Apparently one was supposed to store the GSource as returned by
* G_main_add_ll_cluster(), and g_source_destroy() that "by hand".
*
* But no-one ever did this, not even in the old hb code when this was
* introduced.
*
* Note that fsa_cluster_conn was set as an "alias" to cluster->hb_conn
* in do_ha_control() right after crm_cluster_connect(), and only
* happens to still point at that object, because do_ha_control() does
* not reset it to NULL after crm_cluster_disconnect() above does
* reset cluster->hb_conn to NULL.
* Not sure if that's something to cleanup, too.
*
* I'll try to fix this up in heartbeat proper, so ->delete
* will actually remove, and destroy, and unref, and free this thing.
* Doing so after g_main_loop_quit() is valid with both old,
* and eventually fixed heartbeat.
*
* If we introduce the "by hand" destroy/remove/unref,
* this may break again once heartbeat is fixed :-(
*
* -- Lars Ellenberg
*/
if (fsa_cluster_conn) {
crm_trace("Deleting heartbeat api object");
fsa_cluster_conn->llc_ops->delete(fsa_cluster_conn);
fsa_cluster_conn = NULL;
}
#endif
/* Won't do anything yet, since we're inside it now */
g_main_loop_unref(mloop);
crm_trace("Done %d", rc);
}
/* Graceful */
return rc;
}
/* A_EXIT_0, A_EXIT_1 */
void
do_exit(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
int exit_code = pcmk_ok;
int log_level = LOG_INFO;
const char *exit_type = "gracefully";
if (action & A_EXIT_1) {
/* exit_code = pcmk_err_generic; */
log_level = LOG_ERR;
exit_type = "forcefully";
exit_code = pcmk_err_generic;
}
verify_stopped(cur_state, LOG_ERR);
do_crm_log(log_level, "Performing %s - %s exiting the CRMd",
fsa_action2string(action), exit_type);
crm_info("[%s] stopped (%d)", crm_system_name, exit_code);
crmd_exit(exit_code);
}
static void sigpipe_ignore(int nsig) { return; }
/* A_STARTUP */
void
do_startup(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
int was_error = 0;
crm_debug("Registering Signal Handlers");
mainloop_add_signal(SIGTERM, crm_shutdown);
mainloop_add_signal(SIGPIPE, sigpipe_ignore);
fsa_source = mainloop_add_trigger(G_PRIORITY_HIGH, crm_fsa_trigger, NULL);
config_read = mainloop_add_trigger(G_PRIORITY_HIGH, crm_read_options, NULL);
transition_trigger = mainloop_add_trigger(G_PRIORITY_LOW, te_graph_trigger, NULL);
crm_debug("Creating CIB and LRM objects");
fsa_cib_conn = cib_new();
lrm_state_init_local();
/* set up the timers */
transition_timer = calloc(1, sizeof(fsa_timer_t));
integration_timer = calloc(1, sizeof(fsa_timer_t));
finalization_timer = calloc(1, sizeof(fsa_timer_t));
election_trigger = calloc(1, sizeof(fsa_timer_t));
shutdown_escalation_timer = calloc(1, sizeof(fsa_timer_t));
wait_timer = calloc(1, sizeof(fsa_timer_t));
recheck_timer = calloc(1, sizeof(fsa_timer_t));
if (election_trigger != NULL) {
election_trigger->source_id = 0;
election_trigger->period_ms = -1;
election_trigger->fsa_input = I_DC_TIMEOUT;
election_trigger->callback = crm_timer_popped;
election_trigger->repeat = FALSE;
} else {
was_error = TRUE;
}
if (transition_timer != NULL) {
transition_timer->source_id = 0;
transition_timer->period_ms = -1;
transition_timer->fsa_input = I_PE_CALC;
transition_timer->callback = crm_timer_popped;
transition_timer->repeat = FALSE;
} else {
was_error = TRUE;
}
if (integration_timer != NULL) {
integration_timer->source_id = 0;
integration_timer->period_ms = -1;
integration_timer->fsa_input = I_INTEGRATED;
integration_timer->callback = crm_timer_popped;
integration_timer->repeat = FALSE;
} else {
was_error = TRUE;
}
if (finalization_timer != NULL) {
finalization_timer->source_id = 0;
finalization_timer->period_ms = -1;
finalization_timer->fsa_input = I_FINALIZED;
finalization_timer->callback = crm_timer_popped;
finalization_timer->repeat = FALSE;
/* for possible enabling... a bug in the join protocol left
* a slave in S_PENDING while we think its in S_NOT_DC
*
* raising I_FINALIZED put us into a transition loop which is
* never resolved.
* in this loop we continually send probes which the node
* NACK's because its in S_PENDING
*
* if we have nodes where heartbeat is active but the
* CRM is not... then this will be handled in the
* integration phase
*/
finalization_timer->fsa_input = I_ELECTION;
} else {
was_error = TRUE;
}
if (shutdown_escalation_timer != NULL) {
shutdown_escalation_timer->source_id = 0;
shutdown_escalation_timer->period_ms = -1;
shutdown_escalation_timer->fsa_input = I_STOP;
shutdown_escalation_timer->callback = crm_timer_popped;
shutdown_escalation_timer->repeat = FALSE;
} else {
was_error = TRUE;
}
if (wait_timer != NULL) {
wait_timer->source_id = 0;
wait_timer->period_ms = 2000;
wait_timer->fsa_input = I_NULL;
wait_timer->callback = crm_timer_popped;
wait_timer->repeat = FALSE;
} else {
was_error = TRUE;
}
if (recheck_timer != NULL) {
recheck_timer->source_id = 0;
recheck_timer->period_ms = -1;
recheck_timer->fsa_input = I_PE_CALC;
recheck_timer->callback = crm_timer_popped;
recheck_timer->repeat = FALSE;
} else {
was_error = TRUE;
}
/* set up the sub systems */
cib_subsystem = calloc(1, sizeof(struct crm_subsystem_s));
te_subsystem = calloc(1, sizeof(struct crm_subsystem_s));
pe_subsystem = calloc(1, sizeof(struct crm_subsystem_s));
if (cib_subsystem != NULL) {
cib_subsystem->pid = -1;
cib_subsystem->name = CRM_SYSTEM_CIB;
cib_subsystem->flag_connected = R_CIB_CONNECTED;
cib_subsystem->flag_required = R_CIB_REQUIRED;
} else {
was_error = TRUE;
}
if (te_subsystem != NULL) {
te_subsystem->pid = -1;
te_subsystem->name = CRM_SYSTEM_TENGINE;
te_subsystem->flag_connected = R_TE_CONNECTED;
te_subsystem->flag_required = R_TE_REQUIRED;
} else {
was_error = TRUE;
}
if (pe_subsystem != NULL) {
pe_subsystem->pid = -1;
pe_subsystem->path = CRM_DAEMON_DIR;
pe_subsystem->name = CRM_SYSTEM_PENGINE;
pe_subsystem->command = CRM_DAEMON_DIR "/" CRM_SYSTEM_PENGINE;
pe_subsystem->args = NULL;
pe_subsystem->flag_connected = R_PE_CONNECTED;
pe_subsystem->flag_required = R_PE_REQUIRED;
} else {
was_error = TRUE;
}
if (was_error == FALSE && need_spawn_pengine_from_crmd()) {
if (start_subsystem(pe_subsystem) == FALSE) {
was_error = TRUE;
}
}
if (was_error) {
register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
}
}
static int32_t
crmd_ipc_accept(qb_ipcs_connection_t * c, uid_t uid, gid_t gid)
{
crm_trace("Connection %p", c);
if (crm_client_new(c, uid, gid) == NULL) {
return -EIO;
}
return 0;
}
static void
crmd_ipc_created(qb_ipcs_connection_t * c)
{
crm_trace("Connection %p", c);
}
static int32_t
crmd_ipc_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
{
uint32_t id = 0;
uint32_t flags = 0;
crm_client_t *client = crm_client_get(c);
xmlNode *msg = crm_ipcs_recv(client, data, size, &id, &flags);
crm_trace("Invoked: %s", crm_client_name(client));
crm_ipcs_send_ack(client, id, flags, "ack", __FUNCTION__, __LINE__);
if (msg == NULL) {
return 0;
}
#if ENABLE_ACL
CRM_ASSERT(client->user != NULL);
crm_acl_get_set_user(msg, F_CRM_USER, client->user);
#endif
crm_trace("Processing msg from %s", crm_client_name(client));
crm_log_xml_trace(msg, "CRMd[inbound]");
crm_xml_add(msg, F_CRM_SYS_FROM, client->id);
if (crmd_authorize_message(msg, client, NULL)) {
route_message(C_IPC_MESSAGE, msg);
}
trigger_fsa(fsa_source);
free_xml(msg);
return 0;
}
static int32_t
crmd_ipc_closed(qb_ipcs_connection_t * c)
{
crm_client_t *client = crm_client_get(c);
struct crm_subsystem_s *the_subsystem = NULL;
if (client == NULL) {
return 0;
}
crm_trace("Connection %p", c);
if (client->userdata == NULL) {
crm_trace("Client hadn't registered with us yet");
} else if (strcasecmp(CRM_SYSTEM_PENGINE, client->userdata) == 0) {
the_subsystem = pe_subsystem;
} else if (strcasecmp(CRM_SYSTEM_TENGINE, client->userdata) == 0) {
the_subsystem = te_subsystem;
} else if (strcasecmp(CRM_SYSTEM_CIB, client->userdata) == 0) {
the_subsystem = cib_subsystem;
}
if (the_subsystem != NULL) {
the_subsystem->source = NULL;
the_subsystem->client = NULL;
crm_info("Received HUP from %s:[%d]", the_subsystem->name, the_subsystem->pid);
} else {
/* else that was a transient client */
crm_trace("Received HUP from transient client");
}
crm_trace("Disconnecting client %s (%p)", crm_client_name(client), client);
free(client->userdata);
crm_client_destroy(client);
trigger_fsa(fsa_source);
return 0;
}
static void
crmd_ipc_destroy(qb_ipcs_connection_t * c)
{
crm_trace("Connection %p", c);
crmd_ipc_closed(c);
}
/* A_STOP */
void
do_stop(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
crm_trace("Closing IPC server");
mainloop_del_ipc_server(ipcs); ipcs = NULL;
register_fsa_input(C_FSA_INTERNAL, I_TERMINATE, NULL);
}
/* A_STARTED */
void
do_started(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
static struct qb_ipcs_service_handlers crmd_callbacks = {
.connection_accept = crmd_ipc_accept,
.connection_created = crmd_ipc_created,
.msg_process = crmd_ipc_dispatch,
.connection_closed = crmd_ipc_closed,
.connection_destroyed = crmd_ipc_destroy
};
if (cur_state != S_STARTING) {
crm_err("Start cancelled... %s", fsa_state2string(cur_state));
return;
} else if (is_set(fsa_input_register, R_MEMBERSHIP) == FALSE) {
crm_info("Delaying start, no membership data (%.16llx)", R_MEMBERSHIP);
crmd_fsa_stall(TRUE);
return;
} else if (is_set(fsa_input_register, R_LRM_CONNECTED) == FALSE) {
crm_info("Delaying start, LRM not connected (%.16llx)", R_LRM_CONNECTED);
crmd_fsa_stall(TRUE);
return;
} else if (is_set(fsa_input_register, R_CIB_CONNECTED) == FALSE) {
crm_info("Delaying start, CIB not connected (%.16llx)", R_CIB_CONNECTED);
crmd_fsa_stall(TRUE);
return;
} else if (is_set(fsa_input_register, R_READ_CONFIG) == FALSE) {
crm_info("Delaying start, Config not read (%.16llx)", R_READ_CONFIG);
crmd_fsa_stall(TRUE);
return;
} else if (is_set(fsa_input_register, R_PEER_DATA) == FALSE) {
/* try reading from HA */
crm_info("Delaying start, No peer data (%.16llx)", R_PEER_DATA);
#if SUPPORT_HEARTBEAT
if (is_heartbeat_cluster()) {
HA_Message *msg = NULL;
crm_trace("Looking for a HA message");
msg = fsa_cluster_conn->llc_ops->readmsg(fsa_cluster_conn, 0);
if (msg != NULL) {
crm_trace("There was a HA message");
ha_msg_del(msg);
}
}
#endif
crmd_fsa_stall(TRUE);
return;
}
crm_debug("Init server comms");
ipcs = crmd_ipc_server_init(&crmd_callbacks);
if (ipcs == NULL) {
crm_err("Failed to create IPC server: shutting down and inhibiting respawn");
register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
}
if (stonith_reconnect == NULL) {
int dummy;
stonith_reconnect = mainloop_add_trigger(G_PRIORITY_LOW, te_connect_stonith, &dummy);
}
set_bit(fsa_input_register, R_ST_REQUIRED);
mainloop_set_trigger(stonith_reconnect);
crm_notice("The local CRM is operational");
clear_bit(fsa_input_register, R_STARTING);
register_fsa_input(msg_data->fsa_cause, I_PENDING, NULL);
}
/* A_RECOVER */
void
do_recover(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state, enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
set_bit(fsa_input_register, R_IN_RECOVERY);
crm_warn("Fast-tracking shutdown in response to errors");
register_fsa_input(C_FSA_INTERNAL, I_TERMINATE, NULL);
}
/* *INDENT-OFF* */
pe_cluster_option crmd_opts[] = {
/* name, old-name, validate, default, description */
{ "dc-version", NULL, "string", NULL, "none", NULL, "Version of Pacemaker on the cluster's DC.", "Includes the hash which identifies the exact Mercurial changeset it was built from. Used for diagnostic purposes." },
{ "cluster-infrastructure", NULL, "string", NULL, "heartbeat", NULL, "The messaging stack on which Pacemaker is currently running.", "Used for informational and diagnostic purposes." },
{ XML_CONFIG_ATTR_DC_DEADTIME, "dc_deadtime", "time", NULL, "20s", &check_time, "How long to wait for a response from other nodes during startup.", "The \"correct\" value will depend on the speed/load of your network and the type of switches used." },
{ XML_CONFIG_ATTR_RECHECK, "cluster_recheck_interval", "time",
"Zero disables polling. Positive values are an interval in seconds (unless other SI units are specified. eg. 5min)", "15min", &check_timer,
"Polling interval for time based changes to options, resource parameters and constraints.",
"The Cluster is primarily event driven, however the configuration can have elements that change based on time."
" To ensure these changes take effect, we can optionally poll the cluster's status for changes." },
{ "load-threshold", NULL, "percentage", NULL, "80%", &check_utilization,
"The maximum amount of system resources that should be used by nodes in the cluster",
"The cluster will slow down its recovery process when the amount of system resources used"
" (currently CPU) approaches this limit", },
{ "node-action-limit", NULL, "integer", NULL, "0", &check_number,
"The maximum number of jobs that can be scheduled per node. Defaults to 2x cores"},
{ XML_CONFIG_ATTR_ELECTION_FAIL, "election_timeout", "time", NULL, "2min", &check_timer, "*** Advanced Use Only ***.", "If need to adjust this value, it probably indicates the presence of a bug." },
{ XML_CONFIG_ATTR_FORCE_QUIT, "shutdown_escalation", "time", NULL, "20min", &check_timer, "*** Advanced Use Only ***.", "If need to adjust this value, it probably indicates the presence of a bug." },
{ "crmd-integration-timeout", NULL, "time", NULL, "3min", &check_timer, "*** Advanced Use Only ***.", "If need to adjust this value, it probably indicates the presence of a bug." },
{ "crmd-finalization-timeout", NULL, "time", NULL, "30min", &check_timer, "*** Advanced Use Only ***.", "If you need to adjust this value, it probably indicates the presence of a bug." },
{ "crmd-transition-delay", NULL, "time", NULL, "0s", &check_timer, "*** Advanced Use Only ***\nEnabling this option will slow down cluster recovery under all conditions", "Delay cluster recovery for the configured interval to allow for additional/related events to occur.\nUseful if your configuration is sensitive to the order in which ping updates arrive." },
{ "stonith-watchdog-timeout", NULL, "time", NULL, NULL, &check_timer,
"How long to wait before we can assume nodes are safely down", NULL },
{ "no-quorum-policy", "no_quorum_policy", "enum", "stop, freeze, ignore, suicide", "stop", &check_quorum, NULL, NULL },
#if SUPPORT_PLUGIN
{ XML_ATTR_EXPECTED_VOTES, NULL, "integer", NULL, "2", &check_number, "The number of nodes expected to be in the cluster", "Used to calculate quorum in openais based clusters." },
#endif
};
/* *INDENT-ON* */
void
crmd_metadata(void)
{
config_metadata("CRM Daemon", "1.0",
"CRM Daemon Options",
"This is a fake resource that details the options that can be configured for the CRM Daemon.",
crmd_opts, DIMOF(crmd_opts));
}
static void
verify_crmd_options(GHashTable * options)
{
verify_all_options(options, crmd_opts, DIMOF(crmd_opts));
}
static const char *
crmd_pref(GHashTable * options, const char *name)
{
return get_cluster_pref(options, crmd_opts, DIMOF(crmd_opts), name);
}
static void
config_query_callback(xmlNode * msg, int call_id, int rc, xmlNode * output, void *user_data)
{
const char *value = NULL;
GHashTable *config_hash = NULL;
crm_time_t *now = crm_time_new(NULL);
long st_timeout = 0;
long sbd_timeout = 0;
if (rc != pcmk_ok) {
fsa_data_t *msg_data = NULL;
crm_err("Local CIB query resulted in an error: %s", pcmk_strerror(rc));
register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
if (rc == -EACCES || rc == -pcmk_err_schema_validation) {
crm_err("The cluster is mis-configured - shutting down and staying down");
set_bit(fsa_input_register, R_STAYDOWN);
}
goto bail;
}
crm_debug("Call %d : Parsing CIB options", call_id);
config_hash =
g_hash_table_new_full(crm_str_hash, g_str_equal, g_hash_destroy_str, g_hash_destroy_str);
unpack_instance_attributes(output, output, XML_CIB_TAG_PROPSET, NULL, config_hash,
CIB_OPTIONS_FIRST, FALSE, now);
verify_crmd_options(config_hash);
value = crmd_pref(config_hash, XML_CONFIG_ATTR_DC_DEADTIME);
election_trigger->period_ms = crm_get_msec(value);
value = crmd_pref(config_hash, "node-action-limit"); /* Also checks migration-limit */
throttle_update_job_max(value);
value = crmd_pref(config_hash, "load-threshold");
if(value) {
throttle_load_target = strtof(value, NULL) / 100;
}
value = getenv("SBD_WATCHDOG_TIMEOUT");
sbd_timeout = crm_get_msec(value);
value = crmd_pref(config_hash, "stonith-watchdog-timeout");
st_timeout = crm_get_msec(value);
if(st_timeout > 0 && !daemon_option_enabled(crm_system_name, "watchdog")) {
do_crm_log_always(LOG_EMERG, "Shutting down pacemaker, no watchdog device configured");
crmd_exit(DAEMON_RESPAWN_STOP);
} else if(!daemon_option_enabled(crm_system_name, "watchdog")) {
crm_trace("Watchdog disabled");
} else if(value == NULL && sbd_timeout > 0) {
char *timeout = NULL;
st_timeout = 2 * sbd_timeout / 1000;
timeout = crm_strdup_printf("%lds", st_timeout);
crm_notice("Setting stonith-watchdog-timeout=%s", timeout);
update_attr_delegate(fsa_cib_conn, cib_none, XML_CIB_TAG_CRMCONFIG, NULL, NULL, NULL, NULL,
"stonith-watchdog-timeout", timeout, FALSE, NULL, NULL);
free(timeout);
} else if(st_timeout <= 0) {
crm_notice("Watchdog enabled but stonith-watchdog-timeout is disabled");
} else if(st_timeout < sbd_timeout) {
do_crm_log_always(LOG_EMERG, "Shutting down pacemaker, stonith-watchdog-timeout (%ldms) is too short (must be greater than %ldms)",
st_timeout, sbd_timeout);
crmd_exit(DAEMON_RESPAWN_STOP);
}
value = crmd_pref(config_hash, "no-quorum-policy");
if (safe_str_eq(value, "suicide") && daemon_option_enabled(crm_system_name, "watchdog")) {
no_quorum_suicide_escalation = TRUE;
}
value = crmd_pref(config_hash, XML_CONFIG_ATTR_FORCE_QUIT);
shutdown_escalation_timer->period_ms = crm_get_msec(value);
/* How long to declare an election over - even if not everyone voted */
crm_debug("Shutdown escalation occurs after: %dms", shutdown_escalation_timer->period_ms);
value = crmd_pref(config_hash, XML_CONFIG_ATTR_ELECTION_FAIL);
election_timeout_set_period(fsa_election, crm_get_msec(value));
value = crmd_pref(config_hash, XML_CONFIG_ATTR_RECHECK);
recheck_timer->period_ms = crm_get_msec(value);
crm_debug("Checking for expired actions every %dms", recheck_timer->period_ms);
value = crmd_pref(config_hash, "crmd-transition-delay");
transition_timer->period_ms = crm_get_msec(value);
value = crmd_pref(config_hash, "crmd-integration-timeout");
integration_timer->period_ms = crm_get_msec(value);
value = crmd_pref(config_hash, "crmd-finalization-timeout");
finalization_timer->period_ms = crm_get_msec(value);
#if SUPPORT_COROSYNC
if (is_classic_ais_cluster()) {
value = crmd_pref(config_hash, XML_ATTR_EXPECTED_VOTES);
crm_debug("Sending expected-votes=%s to corosync", value);
send_cluster_text(crm_class_quorum, value, TRUE, NULL, crm_msg_ais);
}
#endif
free(fsa_cluster_name);
fsa_cluster_name = NULL;
value = g_hash_table_lookup(config_hash, "cluster-name");
if (value) {
fsa_cluster_name = strdup(value);
}
set_bit(fsa_input_register, R_READ_CONFIG);
crm_trace("Triggering FSA: %s", __FUNCTION__);
mainloop_set_trigger(fsa_source);
g_hash_table_destroy(config_hash);
bail:
crm_time_free(now);
}
gboolean
crm_read_options(gpointer user_data)
{
int call_id =
fsa_cib_conn->cmds->query(fsa_cib_conn, XML_CIB_TAG_CRMCONFIG, NULL, cib_scope_local);
fsa_register_cib_callback(call_id, FALSE, NULL, config_query_callback);
crm_trace("Querying the CIB... call %d", call_id);
return TRUE;
}
/* A_READCONFIG */
void
do_read_config(long long action,
enum crmd_fsa_cause cause,
enum crmd_fsa_state cur_state,
enum crmd_fsa_input current_input, fsa_data_t * msg_data)
{
throttle_init();
mainloop_set_trigger(config_read);
}
void
crm_shutdown(int nsig)
{
if (crmd_mainloop != NULL && g_main_is_running(crmd_mainloop)) {
if (is_set(fsa_input_register, R_SHUTDOWN)) {
crm_err("Escalating the shutdown");
register_fsa_input_before(C_SHUTDOWN, I_ERROR, NULL);
} else {
set_bit(fsa_input_register, R_SHUTDOWN);
register_fsa_input(C_SHUTDOWN, I_SHUTDOWN, NULL);
if (shutdown_escalation_timer->period_ms < 1) {
const char *value = crmd_pref(NULL, XML_CONFIG_ATTR_FORCE_QUIT);
int msec = crm_get_msec(value);
crm_debug("Using default shutdown escalation: %dms", msec);
shutdown_escalation_timer->period_ms = msec;
}
/* cant rely on this... */
crm_notice("Requesting shutdown, upper limit is %dms",
shutdown_escalation_timer->period_ms);
crm_timer_start(shutdown_escalation_timer);
}
} else {
crm_info("exit from shutdown");
crmd_exit(pcmk_ok);
}
}
diff --git a/fencing/main.c b/fencing/main.c
index 12237b6a76..0069c9d62b 100644
--- a/fencing/main.c
+++ b/fencing/main.c
@@ -1,1476 +1,1479 @@
/*
* Copyright (C) 2009 Andrew Beekhof <andrew@beekhof.net>
*
* 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 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <sys/param.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/utsname.h>
#include <stdlib.h>
#include <errno.h>
#include <fcntl.h>
#include <crm/crm.h>
#include <crm/msg_xml.h>
#include <crm/common/ipc.h>
#include <crm/common/ipcs.h>
#include <crm/cluster/internal.h>
#include <crm/stonith-ng.h>
#include <crm/fencing/internal.h>
#include <crm/common/xml.h>
#include <crm/common/mainloop.h>
#include <crm/cib/internal.h>
#include <crm/pengine/status.h>
#include <allocate.h>
#include <internal.h>
#include <standalone_config.h>
char *stonith_our_uname = NULL;
char *stonith_our_uuid = NULL;
long stonith_watchdog_timeout_ms = 0;
GMainLoop *mainloop = NULL;
gboolean stand_alone = FALSE;
gboolean no_cib_connect = FALSE;
gboolean stonith_shutdown_flag = FALSE;
qb_ipcs_service_t *ipcs = NULL;
xmlNode *local_cib = NULL;
static cib_t *cib_api = NULL;
static void *cib_library = NULL;
static void stonith_shutdown(int nsig);
static void stonith_cleanup(void);
static int32_t
st_ipc_accept(qb_ipcs_connection_t * c, uid_t uid, gid_t gid)
{
if (stonith_shutdown_flag) {
crm_info("Ignoring new client [%d] during shutdown", crm_ipcs_client_pid(c));
return -EPERM;
}
if (crm_client_new(c, uid, gid) == NULL) {
return -EIO;
}
return 0;
}
static void
st_ipc_created(qb_ipcs_connection_t * c)
{
crm_trace("Connection created for %p", c);
}
/* Exit code means? */
static int32_t
st_ipc_dispatch(qb_ipcs_connection_t * qbc, void *data, size_t size)
{
uint32_t id = 0;
uint32_t flags = 0;
int call_options = 0;
xmlNode *request = NULL;
crm_client_t *c = crm_client_get(qbc);
const char *op = NULL;
if (c == NULL) {
crm_info("Invalid client: %p", qbc);
return 0;
}
request = crm_ipcs_recv(c, data, size, &id, &flags);
if (request == NULL) {
crm_ipcs_send_ack(c, id, flags, "nack", __FUNCTION__, __LINE__);
return 0;
}
op = crm_element_value(request, F_CRM_TASK);
if(safe_str_eq(op, CRM_OP_RM_NODE_CACHE)) {
crm_xml_add(request, F_TYPE, T_STONITH_NG);
crm_xml_add(request, F_STONITH_OPERATION, op);
crm_xml_add(request, F_STONITH_CLIENTID, c->id);
crm_xml_add(request, F_STONITH_CLIENTNAME, crm_client_name(c));
crm_xml_add(request, F_STONITH_CLIENTNODE, stonith_our_uname);
send_cluster_message(NULL, crm_msg_stonith_ng, request, FALSE);
free_xml(request);
return 0;
}
if (c->name == NULL) {
const char *value = crm_element_value(request, F_STONITH_CLIENTNAME);
if (value == NULL) {
value = "unknown";
}
c->name = crm_strdup_printf("%s.%u", value, c->pid);
}
crm_element_value_int(request, F_STONITH_CALLOPTS, &call_options);
crm_trace("Flags %u/%u for command %u from %s", flags, call_options, id, crm_client_name(c));
if (is_set(call_options, st_opt_sync_call)) {
CRM_ASSERT(flags & crm_ipc_client_response);
CRM_LOG_ASSERT(c->request_id == 0); /* This means the client has two synchronous events in-flight */
c->request_id = id; /* Reply only to the last one */
}
crm_xml_add(request, F_STONITH_CLIENTID, c->id);
crm_xml_add(request, F_STONITH_CLIENTNAME, crm_client_name(c));
crm_xml_add(request, F_STONITH_CLIENTNODE, stonith_our_uname);
crm_log_xml_trace(request, "Client[inbound]");
stonith_command(c, id, flags, request, NULL);
free_xml(request);
return 0;
}
/* Error code means? */
static int32_t
st_ipc_closed(qb_ipcs_connection_t * c)
{
crm_client_t *client = crm_client_get(c);
if (client == NULL) {
return 0;
}
crm_trace("Connection %p closed", c);
crm_client_destroy(client);
/* 0 means: yes, go ahead and destroy the connection */
return 0;
}
static void
st_ipc_destroy(qb_ipcs_connection_t * c)
{
crm_trace("Connection %p destroyed", c);
st_ipc_closed(c);
}
static void
stonith_peer_callback(xmlNode * msg, void *private_data)
{
const char *remote_peer = crm_element_value(msg, F_ORIG);
const char *op = crm_element_value(msg, F_STONITH_OPERATION);
if (crm_str_eq(op, "poke", TRUE)) {
return;
}
crm_log_xml_trace(msg, "Peer[inbound]");
stonith_command(NULL, 0, 0, msg, remote_peer);
}
#if SUPPORT_HEARTBEAT
static void
stonith_peer_hb_callback(HA_Message * msg, void *private_data)
{
xmlNode *xml = convert_ha_message(NULL, msg, __FUNCTION__);
stonith_peer_callback(xml, private_data);
free_xml(xml);
}
static void
stonith_peer_hb_destroy(gpointer user_data)
{
if (stonith_shutdown_flag) {
crm_info("Heartbeat disconnection complete... exiting");
} else {
crm_err("Heartbeat connection lost! Exiting.");
}
stonith_shutdown(0);
}
#endif
#if SUPPORT_COROSYNC
static void
stonith_peer_ais_callback(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;
xmlNode *xml = NULL;
const char *from = NULL;
char *data = pcmk_message_common_cs(handle, nodeid, pid, msg, &kind, &from);
if(data == NULL) {
return;
}
if (kind == crm_class_cluster) {
xml = string2xml(data);
if (xml == NULL) {
crm_err("Invalid XML: '%.120s'", data);
free(data);
return;
}
crm_xml_add(xml, F_ORIG, from);
/* crm_xml_add_int(xml, F_SEQ, wrapper->id); */
stonith_peer_callback(xml, NULL);
}
free_xml(xml);
free(data);
return;
}
static void
stonith_peer_cs_destroy(gpointer user_data)
{
crm_err("Corosync connection terminated");
stonith_shutdown(0);
}
#endif
void
do_local_reply(xmlNode * notify_src, const char *client_id, gboolean sync_reply, gboolean from_peer)
{
/* send callback to originating child */
crm_client_t *client_obj = NULL;
int local_rc = pcmk_ok;
crm_trace("Sending response");
client_obj = crm_client_get_by_id(client_id);
crm_trace("Sending callback to request originator");
if (client_obj == NULL) {
local_rc = -1;
crm_trace("No client to sent the response to. F_STONITH_CLIENTID not set.");
} else {
int rid = 0;
if (sync_reply) {
CRM_LOG_ASSERT(client_obj->request_id);
rid = client_obj->request_id;
client_obj->request_id = 0;
crm_trace("Sending response %d to %s %s",
rid, client_obj->name, from_peer ? "(originator of delegated request)" : "");
} else {
crm_trace("Sending an event to %s %s",
client_obj->name, from_peer ? "(originator of delegated request)" : "");
}
local_rc = crm_ipcs_send(client_obj, rid, notify_src, sync_reply?crm_ipc_flags_none:crm_ipc_server_event);
}
if (local_rc < pcmk_ok && client_obj != NULL) {
crm_warn("%sSync reply to %s failed: %s",
sync_reply ? "" : "A-",
client_obj ? client_obj->name : "<unknown>", pcmk_strerror(local_rc));
}
}
long long
get_stonith_flag(const char *name)
{
if (safe_str_eq(name, T_STONITH_NOTIFY_FENCE)) {
return 0x01;
} else if (safe_str_eq(name, STONITH_OP_DEVICE_ADD)) {
return 0x04;
} else if (safe_str_eq(name, STONITH_OP_DEVICE_DEL)) {
return 0x10;
}
return 0;
}
static void
stonith_notify_client(gpointer key, gpointer value, gpointer user_data)
{
xmlNode *update_msg = user_data;
crm_client_t *client = value;
const char *type = NULL;
CRM_CHECK(client != NULL, return);
CRM_CHECK(update_msg != NULL, return);
type = crm_element_value(update_msg, F_SUBTYPE);
CRM_CHECK(type != NULL, crm_log_xml_err(update_msg, "notify"); return);
if (client->ipcs == NULL) {
crm_trace("Skipping client with NULL channel");
return;
}
if (client->options & get_stonith_flag(type)) {
int rc = crm_ipcs_send(client, 0, update_msg, crm_ipc_server_event | crm_ipc_server_error);
if (rc <= 0) {
crm_warn("%s notification of client %s.%.6s failed: %s (%d)",
type, crm_client_name(client), client->id, pcmk_strerror(rc), rc);
} else {
crm_trace("Sent %s notification to client %s.%.6s", type, crm_client_name(client),
client->id);
}
}
}
void
do_stonith_async_timeout_update(const char *client_id, const char *call_id, int timeout)
{
crm_client_t *client = NULL;
xmlNode *notify_data = NULL;
if (!timeout || !call_id || !client_id) {
return;
}
client = crm_client_get_by_id(client_id);
if (!client) {
return;
}
notify_data = create_xml_node(NULL, T_STONITH_TIMEOUT_VALUE);
crm_xml_add(notify_data, F_TYPE, T_STONITH_TIMEOUT_VALUE);
crm_xml_add(notify_data, F_STONITH_CALLID, call_id);
crm_xml_add_int(notify_data, F_STONITH_TIMEOUT, timeout);
crm_trace("timeout update is %d for client %s and call id %s", timeout, client_id, call_id);
if (client) {
crm_ipcs_send(client, 0, notify_data, crm_ipc_server_event);
}
free_xml(notify_data);
}
void
do_stonith_notify(int options, const char *type, int result, xmlNode * data)
{
/* TODO: Standardize the contents of data */
xmlNode *update_msg = create_xml_node(NULL, "notify");
CRM_CHECK(type != NULL,;);
crm_xml_add(update_msg, F_TYPE, T_STONITH_NOTIFY);
crm_xml_add(update_msg, F_SUBTYPE, type);
crm_xml_add(update_msg, F_STONITH_OPERATION, type);
crm_xml_add_int(update_msg, F_STONITH_RC, result);
if (data != NULL) {
add_message_xml(update_msg, F_STONITH_CALLDATA, data);
}
crm_trace("Notifying clients");
g_hash_table_foreach(client_connections, stonith_notify_client, update_msg);
free_xml(update_msg);
crm_trace("Notify complete");
}
static stonith_key_value_t *
parse_device_list(const char *devices)
{
int lpc = 0;
int max = 0;
int last = 0;
stonith_key_value_t *output = NULL;
if (devices == NULL) {
return output;
}
max = strlen(devices);
for (lpc = 0; lpc <= max; lpc++) {
if (devices[lpc] == ',' || devices[lpc] == 0) {
char *line = NULL;
line = calloc(1, 2 + lpc - last);
snprintf(line, 1 + lpc - last, "%s", devices + last);
output = stonith_key_value_add(output, NULL, line);
free(line);
last = lpc + 1;
}
}
return output;
}
static void
topology_remove_helper(const char *node, int level)
{
int rc;
char *desc = NULL;
xmlNode *data = create_xml_node(NULL, F_STONITH_LEVEL);
xmlNode *notify_data = create_xml_node(NULL, STONITH_OP_LEVEL_DEL);
crm_xml_add(data, F_STONITH_ORIGIN, __FUNCTION__);
crm_xml_add_int(data, XML_ATTR_ID, level);
crm_xml_add(data, F_STONITH_TARGET, node);
rc = stonith_level_remove(data, &desc);
crm_xml_add(notify_data, F_STONITH_DEVICE, desc);
crm_xml_add_int(notify_data, F_STONITH_ACTIVE, g_hash_table_size(topology));
do_stonith_notify(0, STONITH_OP_LEVEL_DEL, rc, notify_data);
free_xml(notify_data);
free_xml(data);
free(desc);
}
static void
topology_register_helper(const char *node, int level, stonith_key_value_t * device_list)
{
int rc;
char *desc = NULL;
xmlNode *notify_data = create_xml_node(NULL, STONITH_OP_LEVEL_ADD);
xmlNode *data = create_level_registration_xml(node, level, device_list);
rc = stonith_level_register(data, &desc);
crm_xml_add(notify_data, F_STONITH_DEVICE, desc);
crm_xml_add_int(notify_data, F_STONITH_ACTIVE, g_hash_table_size(topology));
do_stonith_notify(0, STONITH_OP_LEVEL_ADD, rc, notify_data);
free_xml(notify_data);
free_xml(data);
free(desc);
}
static void
remove_cib_device(xmlXPathObjectPtr xpathObj)
{
int max = numXpathResults(xpathObj), lpc = 0;
for (lpc = 0; lpc < max; lpc++) {
const char *rsc_id = NULL;
const char *standard = NULL;
xmlNode *match = getXpathResult(xpathObj, lpc);
CRM_LOG_ASSERT(match != NULL);
if(match != NULL) {
standard = crm_element_value(match, XML_AGENT_ATTR_CLASS);
}
if (safe_str_neq(standard, "stonith")) {
continue;
}
rsc_id = crm_element_value(match, XML_ATTR_ID);
stonith_device_remove(rsc_id, TRUE);
}
}
static void
handle_topology_change(xmlNode *match, bool remove)
{
CRM_LOG_ASSERT(match != NULL);
if(match) {
int index = 0;
const char *target;
const char *dev_list;
stonith_key_value_t *devices = NULL;
crm_element_value_int(match, XML_ATTR_STONITH_INDEX, &index);
target = crm_element_value(match, XML_ATTR_STONITH_TARGET);
if(target == NULL) {
target = crm_element_value(match, XML_ATTR_STONITH_TARGET_PATTERN);
}
dev_list = crm_element_value(match, XML_ATTR_STONITH_DEVICES);
devices = parse_device_list(dev_list);
crm_trace("Updating %s[%d] (%s) to %s", target, index, ID(match), dev_list);
if(remove) {
topology_remove_helper(target, index);
}
topology_register_helper(target, index, devices);
stonith_key_value_freeall(devices, 1, 1);
}
}
static void
remove_fencing_topology(xmlXPathObjectPtr xpathObj)
{
int max = numXpathResults(xpathObj), lpc = 0;
for (lpc = 0; lpc < max; lpc++) {
xmlNode *match = getXpathResult(xpathObj, lpc);
CRM_LOG_ASSERT(match != NULL);
if (match && crm_element_value(match, XML_DIFF_MARKER)) {
/* Deletion */
int index = 0;
const char *target = crm_element_value(match, XML_ATTR_STONITH_TARGET);
if(target == NULL) {
target = crm_element_value(match, XML_ATTR_STONITH_TARGET_PATTERN);
}
crm_element_value_int(match, XML_ATTR_STONITH_INDEX, &index);
if (target == NULL) {
crm_err("Invalid fencing target in element %s", ID(match));
} else if (index <= 0) {
crm_err("Invalid level for %s in element %s", target, ID(match));
} else {
topology_remove_helper(target, index);
}
/* } else { Deal with modifications during the 'addition' stage */
}
}
}
static void
register_fencing_topology(xmlXPathObjectPtr xpathObj, gboolean force)
{
int max = numXpathResults(xpathObj), lpc = 0;
for (lpc = 0; lpc < max; lpc++) {
xmlNode *match = getXpathResult(xpathObj, lpc);
handle_topology_change(match, TRUE);
}
}
/* Fencing
<diff crm_feature_set="3.0.6">
<diff-removed>
<fencing-topology>
<fencing-level id="f-p1.1" target="pcmk-1" index="1" devices="poison-pill" __crm_diff_marker__="removed:top"/>
<fencing-level id="f-p1.2" target="pcmk-1" index="2" devices="power" __crm_diff_marker__="removed:top"/>
<fencing-level devices="disk,network" id="f-p2.1"/>
</fencing-topology>
</diff-removed>
<diff-added>
<fencing-topology>
<fencing-level id="f-p.1" target="pcmk-1" index="1" devices="poison-pill" __crm_diff_marker__="added:top"/>
<fencing-level id="f-p2.1" target="pcmk-2" index="1" devices="disk,something"/>
<fencing-level id="f-p3.1" target="pcmk-2" index="2" devices="power" __crm_diff_marker__="added:top"/>
</fencing-topology>
</diff-added>
</diff>
*/
static void
fencing_topology_init(xmlNode * msg)
{
xmlXPathObjectPtr xpathObj = NULL;
const char *xpath = "//" XML_TAG_FENCING_LEVEL;
crm_trace("Full topology refresh");
if(topology) {
g_hash_table_destroy(topology);
topology = g_hash_table_new_full(crm_str_hash, g_str_equal, NULL, free_topology_entry);
}
/* Grab everything */
xpathObj = xpath_search(local_cib, xpath);
register_fencing_topology(xpathObj, TRUE);
freeXpathObject(xpathObj);
}
#define rsc_name(x) x->clone_name?x->clone_name:x->id
static void cib_device_update(resource_t *rsc, pe_working_set_t *data_set)
{
node_t *node = NULL;
const char *value = NULL;
const char *rclass = NULL;
node_t *parent = NULL;
gboolean remove = TRUE;
/* TODO: Mark each installed device and remove if untouched when this process finishes */
if(rsc->children) {
GListPtr gIter = NULL;
for (gIter = rsc->children; gIter != NULL; gIter = gIter->next) {
cib_device_update(gIter->data, data_set);
if(rsc->variant == pe_clone || rsc->variant == pe_master) {
crm_trace("Only processing one copy of the clone %s", rsc->id);
break;
}
}
return;
}
rclass = crm_element_value(rsc->xml, XML_AGENT_ATTR_CLASS);
if(safe_str_neq(rclass, "stonith")) {
return;
}
value = g_hash_table_lookup(rsc->meta, XML_RSC_ATTR_TARGET_ROLE);
if(value && strcmp(RSC_STOPPED, value) == 0) {
crm_info("Device %s has been disabled", rsc->id);
goto update_done;
} else if(stonith_our_uname) {
GHashTableIter iter;
g_hash_table_iter_init(&iter, rsc->allowed_nodes);
while (g_hash_table_iter_next(&iter, NULL, (void **)&node)) {
if(node && strcmp(node->details->uname, stonith_our_uname) == 0) {
break;
}
node = NULL;
}
}
if (rsc->parent && rsc->parent->variant == pe_group && stonith_our_uname) {
GHashTableIter iter;
g_hash_table_iter_init(&iter, rsc->parent->allowed_nodes);
while (g_hash_table_iter_next(&iter, NULL, (void **)&parent)) {
if(parent && strcmp(parent->details->uname, stonith_our_uname) == 0) {
break;
}
parent = NULL;
}
}
if(node == NULL) {
GHashTableIter iter;
crm_info("Device %s has been disabled on %s: unknown", rsc->id, stonith_our_uname);
g_hash_table_iter_init(&iter, rsc->allowed_nodes);
while (g_hash_table_iter_next(&iter, NULL, (void **)&node)) {
crm_trace("Available: %s = %d", node->details->uname, node->weight);
}
goto update_done;
} else if(node->weight < 0 || (parent && parent->weight < 0)) {
char *score = score2char((node->weight < 0) ? node->weight : parent->weight);
crm_info("Device %s has been disabled on %s: score=%s", rsc->id, stonith_our_uname, score);
free(score);
goto update_done;
} else {
xmlNode *data;
GHashTableIter gIter;
stonith_key_value_t *params = NULL;
const char *name = NULL;
const char *agent = crm_element_value(rsc->xml, XML_EXPR_ATTR_TYPE);
const char *provider = crm_element_value(rsc->xml, XML_AGENT_ATTR_PROVIDER);
const char *rsc_provides = NULL;
crm_debug("Device %s is allowed on %s: score=%d", rsc->id, stonith_our_uname, node->weight);
get_rsc_attributes(rsc->parameters, rsc, node, data_set);
get_meta_attributes(rsc->meta, rsc, node, data_set);
rsc_provides = g_hash_table_lookup(rsc->meta, XML_RSC_ATTR_PROVIDES);
g_hash_table_iter_init(&gIter, rsc->parameters);
while (g_hash_table_iter_next(&gIter, (gpointer *) & name, (gpointer *) & value)) {
if (!name || !value) {
continue;
}
params = stonith_key_value_add(params, name, value);
crm_trace(" %s=%s", name, value);
}
remove = FALSE;
data = create_device_registration_xml(rsc_name(rsc), provider, agent, params, rsc_provides);
stonith_device_register(data, NULL, TRUE);
stonith_key_value_freeall(params, 1, 1);
free_xml(data);
}
update_done:
if(remove && g_hash_table_lookup(device_list, rsc_name(rsc))) {
stonith_device_remove(rsc_name(rsc), TRUE);
}
}
extern xmlNode *do_calculations(pe_working_set_t * data_set, xmlNode * xml_input, crm_time_t * now);
extern node_t *create_node(const char *id, const char *uname, const char *type, const char *score, pe_working_set_t * data_set);
static void
cib_devices_update(void)
{
GListPtr gIter = NULL;
pe_working_set_t data_set;
crm_info("Updating devices to version %s.%s.%s",
crm_element_value(local_cib, XML_ATTR_GENERATION_ADMIN),
crm_element_value(local_cib, XML_ATTR_GENERATION),
crm_element_value(local_cib, XML_ATTR_NUMUPDATES));
set_working_set_defaults(&data_set);
data_set.input = local_cib;
data_set.now = crm_time_new(NULL);
data_set.flags |= pe_flag_quick_location;
data_set.localhost = stonith_our_uname;
cluster_status(&data_set);
do_calculations(&data_set, NULL, NULL);
for (gIter = data_set.resources; gIter != NULL; gIter = gIter->next) {
cib_device_update(gIter->data, &data_set);
}
data_set.input = NULL; /* Wasn't a copy */
cleanup_alloc_calculations(&data_set);
}
static void
update_cib_stonith_devices_v2(const char *event, xmlNode * msg)
{
xmlNode *change = NULL;
char *reason = NULL;
bool needs_update = FALSE;
xmlNode *patchset = get_message_xml(msg, F_CIB_UPDATE_RESULT);
for (change = __xml_first_child(patchset); change != NULL; change = __xml_next(change)) {
const char *op = crm_element_value(change, XML_DIFF_OP);
const char *xpath = crm_element_value(change, XML_DIFF_PATH);
const char *shortpath = NULL;
if(op == NULL || strcmp(op, "move") == 0) {
continue;
} else if(safe_str_eq(op, "delete") && strstr(xpath, XML_CIB_TAG_RESOURCE)) {
const char *rsc_id = NULL;
char *search = NULL;
char *mutable = strdup(xpath);
rsc_id = strstr(mutable, "primitive[@id=\'") + strlen("primitive[@id=\'");
search = strchr(rsc_id, '\'');
search[0] = 0;
stonith_device_remove(rsc_id, TRUE);
free(mutable);
} else if(strstr(xpath, "/"XML_CIB_TAG_RESOURCES)) {
shortpath = strrchr(xpath, '/'); CRM_ASSERT(shortpath);
reason = crm_strdup_printf("%s %s", op, shortpath+1);
needs_update = TRUE;
break;
} else if(strstr(xpath, XML_CONS_TAG_RSC_LOCATION)) {
shortpath = strrchr(xpath, '/'); CRM_ASSERT(shortpath);
reason = crm_strdup_printf("%s %s", op, shortpath+1);
needs_update = TRUE;
break;
}
}
if(needs_update) {
crm_info("Updating device list from the cib: %s", reason);
cib_devices_update();
}
free(reason);
}
static void
update_cib_stonith_devices_v1(const char *event, xmlNode * msg)
{
const char *reason = "none";
gboolean needs_update = FALSE;
xmlXPathObjectPtr xpath_obj = NULL;
/* process new constraints */
xpath_obj = xpath_search(msg, "//" F_CIB_UPDATE_RESULT "//" XML_CONS_TAG_RSC_LOCATION);
if (numXpathResults(xpath_obj) > 0) {
int max = numXpathResults(xpath_obj), lpc = 0;
/* Safest and simplest to always recompute */
needs_update = TRUE;
reason = "new location constraint";
for (lpc = 0; lpc < max; lpc++) {
xmlNode *match = getXpathResult(xpath_obj, lpc);
crm_log_xml_trace(match, "new constraint");
}
}
freeXpathObject(xpath_obj);
/* process deletions */
xpath_obj = xpath_search(msg, "//" F_CIB_UPDATE_RESULT "//" XML_TAG_DIFF_REMOVED "//" XML_CIB_TAG_RESOURCE);
if (numXpathResults(xpath_obj) > 0) {
remove_cib_device(xpath_obj);
}
freeXpathObject(xpath_obj);
/* process additions */
xpath_obj = xpath_search(msg, "//" F_CIB_UPDATE_RESULT "//" XML_TAG_DIFF_ADDED "//" XML_CIB_TAG_RESOURCE);
if (numXpathResults(xpath_obj) > 0) {
int max = numXpathResults(xpath_obj), lpc = 0;
for (lpc = 0; lpc < max; lpc++) {
const char *rsc_id = NULL;
const char *standard = NULL;
xmlNode *match = getXpathResult(xpath_obj, lpc);
rsc_id = crm_element_value(match, XML_ATTR_ID);
standard = crm_element_value(match, XML_AGENT_ATTR_CLASS);
if (safe_str_neq(standard, "stonith")) {
continue;
}
crm_trace("Fencing resource %s was added or modified", rsc_id);
reason = "new resource";
needs_update = TRUE;
}
}
freeXpathObject(xpath_obj);
if(needs_update) {
crm_info("Updating device list from the cib: %s", reason);
cib_devices_update();
}
}
static void
update_cib_stonith_devices(const char *event, xmlNode * msg)
{
int format = 1;
xmlNode *patchset = get_message_xml(msg, F_CIB_UPDATE_RESULT);
CRM_ASSERT(patchset);
crm_element_value_int(patchset, "format", &format);
switch(format) {
case 1:
update_cib_stonith_devices_v1(event, msg);
break;
case 2:
update_cib_stonith_devices_v2(event, msg);
break;
default:
crm_warn("Unknown patch format: %d", format);
}
}
static void
update_fencing_topology(const char *event, xmlNode * msg)
{
int format = 1;
const char *xpath;
xmlXPathObjectPtr xpathObj = NULL;
xmlNode *patchset = get_message_xml(msg, F_CIB_UPDATE_RESULT);
CRM_ASSERT(patchset);
crm_element_value_int(patchset, "format", &format);
if(format == 1) {
/* Process deletions (only) */
xpath = "//" F_CIB_UPDATE_RESULT "//" XML_TAG_DIFF_REMOVED "//" XML_TAG_FENCING_LEVEL;
xpathObj = xpath_search(msg, xpath);
remove_fencing_topology(xpathObj);
freeXpathObject(xpathObj);
/* Process additions and changes */
xpath = "//" F_CIB_UPDATE_RESULT "//" XML_TAG_DIFF_ADDED "//" XML_TAG_FENCING_LEVEL;
xpathObj = xpath_search(msg, xpath);
register_fencing_topology(xpathObj, FALSE);
freeXpathObject(xpathObj);
} else if(format == 2) {
xmlNode *change = NULL;
int add[] = { 0, 0, 0 };
int del[] = { 0, 0, 0 };
xml_patch_versions(patchset, add, del);
for (change = __xml_first_child(patchset); change != NULL; change = __xml_next(change)) {
const char *op = crm_element_value(change, XML_DIFF_OP);
const char *xpath = crm_element_value(change, XML_DIFF_PATH);
if(op == NULL) {
continue;
} else if(strstr(xpath, "/" XML_TAG_FENCING_LEVEL) != NULL) {
/* Change to a specific entry */
crm_trace("Handling %s operation %d.%d.%d for %s", op, add[0], add[1], add[2], xpath);
if(strcmp(op, "move") == 0) {
continue;
} else if(strcmp(op, "create") == 0) {
handle_topology_change(change->children, FALSE);
} else if(strcmp(op, "modify") == 0) {
xmlNode *match = first_named_child(change, XML_DIFF_RESULT);
if(match) {
handle_topology_change(match->children, TRUE);
}
} else if(strcmp(op, "delete") == 0) {
/* Nuclear option, all we have is the path and an id... not enough to remove a specific entry */
crm_info("Re-initializing fencing topology after %s operation %d.%d.%d for %s",
op, add[0], add[1], add[2], xpath);
fencing_topology_init(NULL);
return;
}
} else if (strstr(xpath, "/" XML_TAG_FENCING_TOPOLOGY) != NULL) {
/* Change to the topology in general */
crm_info("Re-initializing fencing topology after top-level %s operation %d.%d.%d for %s",
op, add[0], add[1], add[2], xpath);
fencing_topology_init(NULL);
return;
} else if (strstr(xpath, "/" XML_CIB_TAG_CONFIGURATION)) {
/* Changes to the whole config section, possibly including the topology as a whild */
if(first_named_child(change, XML_TAG_FENCING_TOPOLOGY) == NULL) {
crm_trace("Nothing for us in %s operation %d.%d.%d for %s.",
op, add[0], add[1], add[2], xpath);
} else if(strcmp(op, "delete") == 0 || strcmp(op, "create") == 0) {
crm_info("Re-initializing fencing topology after top-level %s operation %d.%d.%d for %s.",
op, add[0], add[1], add[2], xpath);
fencing_topology_init(NULL);
return;
}
} else {
crm_trace("Nothing for us in %s operation %d.%d.%d for %s",
op, add[0], add[1], add[2], xpath);
}
}
} else {
crm_warn("Unknown patch format: %d", format);
}
}
static bool have_cib_devices = FALSE;
static void
update_cib_cache_cb(const char *event, xmlNode * msg)
{
int rc = pcmk_ok;
xmlNode *stonith_enabled_xml = NULL;
xmlNode *stonith_watchdog_xml = NULL;
const char *stonith_enabled_s = NULL;
static gboolean stonith_enabled_saved = TRUE;
if(!have_cib_devices) {
crm_trace("Skipping updates until we get a full dump");
return;
} else if(msg == NULL) {
crm_trace("Missing %s update", event);
return;
}
/* Maintain a local copy of the CIB so that we have full access to the device definitions and location constraints */
if (local_cib != NULL) {
int rc = pcmk_ok;
xmlNode *patchset = NULL;
crm_element_value_int(msg, F_CIB_RC, &rc);
if (rc != pcmk_ok) {
return;
}
patchset = get_message_xml(msg, F_CIB_UPDATE_RESULT);
xml_log_patchset(LOG_TRACE, "Config update", patchset);
rc = xml_apply_patchset(local_cib, patchset, TRUE);
switch (rc) {
case pcmk_ok:
case -pcmk_err_old_data:
break;
case -pcmk_err_diff_resync:
case -pcmk_err_diff_failed:
crm_notice("[%s] Patch aborted: %s (%d)", event, pcmk_strerror(rc), rc);
free_xml(local_cib);
local_cib = NULL;
break;
default:
crm_warn("[%s] ABORTED: %s (%d)", event, pcmk_strerror(rc), rc);
free_xml(local_cib);
local_cib = NULL;
}
}
if (local_cib == NULL) {
crm_trace("Re-requesting the full cib");
rc = cib_api->cmds->query(cib_api, NULL, &local_cib, cib_scope_local | cib_sync_call);
if(rc != pcmk_ok) {
crm_err("Couldnt retrieve the CIB: %s (%d)", pcmk_strerror(rc), rc);
return;
}
CRM_ASSERT(local_cib != NULL);
stonith_enabled_saved = FALSE; /* Trigger a full refresh below */
}
stonith_enabled_xml = get_xpath_object("//nvpair[@name='stonith-enabled']", local_cib, LOG_TRACE);
if (stonith_enabled_xml) {
stonith_enabled_s = crm_element_value(stonith_enabled_xml, XML_NVPAIR_ATTR_VALUE);
}
if(daemon_option_enabled(crm_system_name, "watchdog")) {
const char *value = NULL;
long timeout_ms = 0;
if(value == NULL) {
stonith_watchdog_xml = get_xpath_object("//nvpair[@name='stonith-watchdog-timeout']", local_cib, LOG_TRACE);
if (stonith_watchdog_xml) {
value = crm_element_value(stonith_watchdog_xml, XML_NVPAIR_ATTR_VALUE);
}
}
if(value) {
timeout_ms = crm_get_msec(value);
}
if(timeout_ms != stonith_watchdog_timeout_ms) {
crm_notice("New watchdog timeout %lds (was %lds)", timeout_ms/1000, stonith_watchdog_timeout_ms/1000);
stonith_watchdog_timeout_ms = timeout_ms;
}
}
if (stonith_enabled_s && crm_is_true(stonith_enabled_s) == FALSE) {
crm_trace("Ignoring cib updates while stonith is disabled");
stonith_enabled_saved = FALSE;
return;
} else if (stonith_enabled_saved == FALSE) {
crm_info("Updating stonith device and topology lists now that stonith is enabled");
stonith_enabled_saved = TRUE;
fencing_topology_init(NULL);
cib_devices_update();
} else {
update_fencing_topology(event, msg);
update_cib_stonith_devices(event, msg);
}
}
static void
init_cib_cache_cb(xmlNode * msg, int call_id, int rc, xmlNode * output, void *user_data)
{
crm_info("Updating device list from the cib: init");
have_cib_devices = TRUE;
local_cib = copy_xml(output);
fencing_topology_init(msg);
cib_devices_update();
}
static void
stonith_shutdown(int nsig)
{
stonith_shutdown_flag = TRUE;
crm_info("Terminating with %d clients", crm_hash_table_size(client_connections));
if (mainloop != NULL && g_main_is_running(mainloop)) {
g_main_quit(mainloop);
} else {
stonith_cleanup();
crm_exit(pcmk_ok);
}
}
static void
cib_connection_destroy(gpointer user_data)
{
if (stonith_shutdown_flag) {
crm_info("Connection to the CIB closed.");
return;
} else {
crm_notice("Connection to the CIB terminated. Shutting down.");
}
if (cib_api) {
cib_api->cmds->signoff(cib_api);
}
stonith_shutdown(0);
}
static void
stonith_cleanup(void)
{
if (cib_api) {
cib_api->cmds->signoff(cib_api);
}
if (ipcs) {
qb_ipcs_destroy(ipcs);
}
crm_peer_destroy();
crm_client_cleanup();
free(stonith_our_uname);
free_xml(local_cib);
}
/* *INDENT-OFF* */
static struct crm_option long_options[] = {
{"stand-alone", 0, 0, 's'},
{"stand-alone-w-cpg", 0, 0, 'c'},
{"logfile", 1, 0, 'l'},
{"verbose", 0, 0, 'V'},
{"version", 0, 0, '$'},
{"help", 0, 0, '?'},
{0, 0, 0, 0}
};
/* *INDENT-ON* */
static void
setup_cib(void)
{
int rc, retries = 0;
static cib_t *(*cib_new_fn) (void) = NULL;
if (cib_new_fn == NULL) {
cib_new_fn = find_library_function(&cib_library, CIB_LIBRARY, "cib_new", TRUE);
}
if (cib_new_fn != NULL) {
cib_api = (*cib_new_fn) ();
}
if (cib_api == NULL) {
crm_err("No connection to the CIB");
return;
}
do {
sleep(retries);
rc = cib_api->cmds->signon(cib_api, CRM_SYSTEM_CRMD, cib_command);
} while (rc == -ENOTCONN && ++retries < 5);
if (rc != pcmk_ok) {
crm_err("Could not connect to the CIB service: %s (%d)", pcmk_strerror(rc), rc);
} else if (pcmk_ok !=
cib_api->cmds->add_notify_callback(cib_api, T_CIB_DIFF_NOTIFY, update_cib_cache_cb)) {
crm_err("Could not set CIB notification callback");
} else {
rc = cib_api->cmds->query(cib_api, NULL, NULL, cib_scope_local);
cib_api->cmds->register_callback(cib_api, rc, 120, FALSE, NULL, "init_cib_cache_cb",
init_cib_cache_cb);
cib_api->cmds->set_connection_dnotify(cib_api, cib_connection_destroy);
crm_notice("Watching for stonith topology changes");
}
}
struct qb_ipcs_service_handlers ipc_callbacks = {
.connection_accept = st_ipc_accept,
.connection_created = st_ipc_created,
.msg_process = st_ipc_dispatch,
.connection_closed = st_ipc_closed,
.connection_destroyed = st_ipc_destroy
};
+/*!
+ * \internal
+ * \brief Callback for peer status changes
+ *
+ * \param[in] type What changed
+ * \param[in] node What peer had the change
+ * \param[in] data Previous value of what changed
+ */
static void
st_peer_update_callback(enum crm_status_type type, crm_node_t * node, const void *data)
{
- xmlNode *query = NULL;
+ if (type == crm_status_uname) {
+ /*
+ * This is a hack until we can send to a nodeid and/or we fix node name lookups
+ * These messages are ignored in stonith_peer_callback()
+ */
+ xmlNode *query = query = create_xml_node(NULL, "stonith_command");
- if (type == crm_status_processes) {
- crm_update_peer_state(__FUNCTION__, node, is_set(node->processes, crm_proc_cpg)?CRM_NODE_MEMBER:CRM_NODE_LOST, 0);
- return;
- }
+ crm_xml_add(query, F_XML_TAGNAME, "stonith_command");
+ crm_xml_add(query, F_TYPE, T_STONITH_NG);
+ crm_xml_add(query, F_STONITH_OPERATION, "poke");
- /*
- * This is a hack until we can send to a nodeid and/or we fix node name lookups
- * These messages are ignored in stonith_peer_callback()
- */
- query = create_xml_node(NULL, "stonith_command");
+ crm_debug("Broadcasting our uname because of node %u", node->id);
+ send_cluster_message(NULL, crm_msg_stonith_ng, query, FALSE);
- crm_xml_add(query, F_XML_TAGNAME, "stonith_command");
- crm_xml_add(query, F_TYPE, T_STONITH_NG);
- crm_xml_add(query, F_STONITH_OPERATION, "poke");
-
- crm_debug("Broadcasting our uname because of node %u", node->id);
- send_cluster_message(NULL, crm_msg_stonith_ng, query, FALSE);
-
- free_xml(query);
+ free_xml(query);
+ }
}
int
main(int argc, char **argv)
{
int flag;
int rc = 0;
int lpc = 0;
int argerr = 0;
int option_index = 0;
crm_cluster_t cluster;
const char *actions[] = { "reboot", "off", "list", "monitor", "status" };
crm_log_preinit("stonith-ng", argc, argv);
crm_set_options(NULL, "mode [options]", long_options,
"Provides a summary of cluster's current state."
"\n\nOutputs varying levels of detail in a number of different formats.\n");
while (1) {
flag = crm_get_option(argc, argv, &option_index);
if (flag == -1) {
break;
}
switch (flag) {
case 'V':
crm_bump_log_level(argc, argv);
break;
case 'l':
crm_add_logfile(optarg);
break;
case 's':
stand_alone = TRUE;
break;
case 'c':
stand_alone = FALSE;
no_cib_connect = TRUE;
break;
case '$':
case '?':
crm_help(flag, EX_OK);
break;
default:
++argerr;
break;
}
}
if (argc - optind == 1 && safe_str_eq("metadata", argv[optind])) {
printf("<?xml version=\"1.0\"?><!DOCTYPE resource-agent SYSTEM \"ra-api-1.dtd\">\n");
printf("<resource-agent name=\"stonithd\">\n");
printf(" <version>1.0</version>\n");
printf
(" <longdesc lang=\"en\">This is a fake resource that details the instance attributes handled by stonithd.</longdesc>\n");
printf(" <shortdesc lang=\"en\">Options available for all stonith resources</shortdesc>\n");
printf(" <parameters>\n");
printf(" <parameter name=\"priority\" unique=\"0\">\n");
printf
(" <shortdesc lang=\"en\">The priority of the stonith resource. Devices are tried in order of highest priority to lowest.</shortdesc>\n");
printf(" <content type=\"integer\" default=\"0\"/>\n");
printf(" </parameter>\n");
printf(" <parameter name=\"%s\" unique=\"0\">\n", STONITH_ATTR_HOSTARG);
printf
(" <shortdesc lang=\"en\">Advanced use only: An alternate parameter to supply instead of 'port'</shortdesc>\n");
printf
(" <longdesc lang=\"en\">Some devices do not support the standard 'port' parameter or may provide additional ones.\n"
"Use this to specify an alternate, device-specific, parameter that should indicate the machine to be fenced.\n"
"A value of 'none' can be used to tell the cluster not to supply any additional parameters.\n"
" </longdesc>\n");
printf(" <content type=\"string\" default=\"port\"/>\n");
printf(" </parameter>\n");
printf(" <parameter name=\"%s\" unique=\"0\">\n", STONITH_ATTR_HOSTMAP);
printf
(" <shortdesc lang=\"en\">A mapping of host names to ports numbers for devices that do not support host names.</shortdesc>\n");
printf
(" <longdesc lang=\"en\">Eg. node1:1;node2:2,3 would tell the cluster to use port 1 for node1 and ports 2 and 3 for node2</longdesc>\n");
printf(" <content type=\"string\" default=\"\"/>\n");
printf(" </parameter>\n");
printf(" <parameter name=\"%s\" unique=\"0\">\n", STONITH_ATTR_HOSTLIST);
printf
(" <shortdesc lang=\"en\">A list of machines controlled by this device (Optional unless %s=static-list).</shortdesc>\n",
STONITH_ATTR_HOSTCHECK);
printf(" <content type=\"string\" default=\"\"/>\n");
printf(" </parameter>\n");
printf(" <parameter name=\"%s\" unique=\"0\">\n", STONITH_ATTR_HOSTCHECK);
printf
(" <shortdesc lang=\"en\">How to determine which machines are controlled by the device.</shortdesc>\n");
printf
(" <longdesc lang=\"en\">Allowed values: dynamic-list (query the device), static-list (check the %s attribute), none (assume every device can fence every machine)</longdesc>\n",
STONITH_ATTR_HOSTLIST);
printf(" <content type=\"string\" default=\"dynamic-list\"/>\n");
printf(" </parameter>\n");
printf(" <parameter name=\"%s\" unique=\"0\">\n", STONITH_ATTR_DELAY_MAX);
printf
(" <shortdesc lang=\"en\">Enable random delay for stonith actions and specify the maximum of random delay</shortdesc>\n");
printf
(" <longdesc lang=\"en\">This prevents double fencing when using slow devices such as sbd.\n"
"Use this to enable random delay for stonith actions and specify the maximum of random delay.</longdesc>\n");
printf(" <content type=\"time\" default=\"0s\"/>\n");
printf(" </parameter>\n");
for (lpc = 0; lpc < DIMOF(actions); lpc++) {
printf(" <parameter name=\"pcmk_%s_action\" unique=\"0\">\n", actions[lpc]);
printf
(" <shortdesc lang=\"en\">Advanced use only: An alternate command to run instead of '%s'</shortdesc>\n",
actions[lpc]);
printf
(" <longdesc lang=\"en\">Some devices do not support the standard commands or may provide additional ones.\n"
"Use this to specify an alternate, device-specific, command that implements the '%s' action.</longdesc>\n",
actions[lpc]);
printf(" <content type=\"string\" default=\"%s\"/>\n", actions[lpc]);
printf(" </parameter>\n");
printf(" <parameter name=\"pcmk_%s_timeout\" unique=\"0\">\n", actions[lpc]);
printf
(" <shortdesc lang=\"en\">Advanced use only: Specify an alternate timeout to use for %s actions instead of stonith-timeout</shortdesc>\n",
actions[lpc]);
printf
(" <longdesc lang=\"en\">Some devices need much more/less time to complete than normal.\n"
"Use this to specify an alternate, device-specific, timeout for '%s' actions.</longdesc>\n",
actions[lpc]);
printf(" <content type=\"time\" default=\"60s\"/>\n");
printf(" </parameter>\n");
printf(" <parameter name=\"pcmk_%s_retries\" unique=\"0\">\n", actions[lpc]);
printf
(" <shortdesc lang=\"en\">Advanced use only: The maximum number of times to retry the '%s' command within the timeout period</shortdesc>\n",
actions[lpc]);
printf(" <longdesc lang=\"en\">Some devices do not support multiple connections."
" Operations may 'fail' if the device is busy with another task so Pacemaker will automatically retry the operation, if there is time remaining."
" Use this option to alter the number of times Pacemaker retries '%s' actions before giving up."
"</longdesc>\n", actions[lpc]);
printf(" <content type=\"integer\" default=\"2\"/>\n");
printf(" </parameter>\n");
}
printf(" </parameters>\n");
printf("</resource-agent>\n");
return 0;
}
if (optind != argc) {
++argerr;
}
if (argerr) {
crm_help('?', EX_USAGE);
}
crm_log_init("stonith-ng", LOG_INFO, TRUE, FALSE, argc, argv, FALSE);
mainloop_add_signal(SIGTERM, stonith_shutdown);
crm_peer_init();
if (stand_alone == FALSE) {
#if SUPPORT_HEARTBEAT
cluster.hb_conn = NULL;
cluster.hb_dispatch = stonith_peer_hb_callback;
cluster.destroy = stonith_peer_hb_destroy;
#endif
if (is_openais_cluster()) {
#if SUPPORT_COROSYNC
cluster.destroy = stonith_peer_cs_destroy;
cluster.cpg.cpg_deliver_fn = stonith_peer_ais_callback;
cluster.cpg.cpg_confchg_fn = pcmk_cpg_membership;
#endif
}
if (crm_cluster_connect(&cluster) == FALSE) {
crm_crit("Cannot sign in to the cluster... terminating");
crm_exit(DAEMON_RESPAWN_STOP);
}
stonith_our_uname = cluster.uname;
stonith_our_uuid = cluster.uuid;
#if SUPPORT_HEARTBEAT
if (is_heartbeat_cluster()) {
/* crm_cluster_connect() registered us for crm_system_name, which
* usually is the only F_TYPE used by the respective sub system.
* Stonith needs to register two additional F_TYPE callbacks,
* because it can :-/ */
if (HA_OK !=
cluster.hb_conn->llc_ops->set_msg_callback(cluster.hb_conn, T_STONITH_NOTIFY,
cluster.hb_dispatch, cluster.hb_conn)) {
crm_crit("Cannot set msg callback %s: %s", T_STONITH_NOTIFY, cluster.hb_conn->llc_ops->errmsg(cluster.hb_conn));
crm_exit(DAEMON_RESPAWN_STOP);
}
if (HA_OK !=
cluster.hb_conn->llc_ops->set_msg_callback(cluster.hb_conn, T_STONITH_TIMEOUT_VALUE,
cluster.hb_dispatch, cluster.hb_conn)) {
crm_crit("Cannot set msg callback %s: %s", T_STONITH_TIMEOUT_VALUE, cluster.hb_conn->llc_ops->errmsg(cluster.hb_conn));
crm_exit(DAEMON_RESPAWN_STOP);
}
}
#endif
if (no_cib_connect == FALSE) {
setup_cib();
}
} else {
stonith_our_uname = strdup("localhost");
}
crm_set_status_callback(&st_peer_update_callback);
device_list = g_hash_table_new_full(crm_str_hash, g_str_equal, NULL, free_device);
topology = g_hash_table_new_full(crm_str_hash, g_str_equal, NULL, free_topology_entry);
if(daemon_option_enabled(crm_system_name, "watchdog")) {
xmlNode *xml;
stonith_key_value_t *params = NULL;
params = stonith_key_value_add(params, STONITH_ATTR_HOSTLIST, stonith_our_uname);
xml = create_device_registration_xml("watchdog", "internal", STONITH_WATCHDOG_AGENT, params, NULL);
stonith_device_register(xml, NULL, FALSE);
stonith_key_value_freeall(params, 1, 1);
free_xml(xml);
}
stonith_ipc_server_init(&ipcs, &ipc_callbacks);
#if SUPPORT_STONITH_CONFIG
if (((stand_alone == TRUE)) && !(standalone_cfg_read_file(STONITH_NG_CONF_FILE))) {
standalone_cfg_commit();
}
#endif
/* Create the mainloop and run it... */
mainloop = g_main_new(FALSE);
crm_info("Starting %s mainloop", crm_system_name);
g_main_run(mainloop);
stonith_cleanup();
#if SUPPORT_HEARTBEAT
if (cluster.hb_conn) {
cluster.hb_conn->llc_ops->delete(cluster.hb_conn);
}
#endif
crm_info("Done");
return crm_exit(rc);
}
diff --git a/include/crm/cluster.h b/include/crm/cluster.h
index e1530a1f2a..20ed829c7e 100644
--- a/include/crm/cluster.h
+++ b/include/crm/cluster.h
@@ -1,232 +1,233 @@
/*
* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
*
* 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 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef CRM_COMMON_CLUSTER__H
# define CRM_COMMON_CLUSTER__H
# include <crm/common/xml.h>
# include <crm/common/util.h>
# if SUPPORT_HEARTBEAT
# include <heartbeat/hb_api.h>
# include <ocf/oc_event.h>
# endif
# if SUPPORT_COROSYNC
# include <corosync/cpg.h>
# endif
extern gboolean crm_have_quorum;
extern GHashTable *crm_peer_cache;
extern GHashTable *crm_remote_peer_cache;
extern unsigned long long crm_peer_seq;
# ifndef CRM_SERVICE
# define CRM_SERVICE PCMK_SERVICE_ID
# endif
/* *INDENT-OFF* */
#define CRM_NODE_LOST "lost"
#define CRM_NODE_MEMBER "member"
#define CRM_NODE_ACTIVE CRM_NODE_MEMBER
#define CRM_NODE_EVICTED "evicted"
enum crm_join_phase
{
crm_join_nack = -1,
crm_join_none = 0,
crm_join_welcomed = 1,
crm_join_integrated = 2,
crm_join_finalized = 3,
crm_join_confirmed = 4,
};
enum crm_node_flags
{
/* node is not a cluster node and should not be considered for cluster membership */
crm_remote_node = 0x0001,
/* This node is a remote node living within a container resource */
crm_remote_container = 0x0002,
/* This node is a bare metal remote-node */
crm_remote_baremetal = 0x0004,
};
/* *INDENT-ON* */
typedef struct crm_peer_node_s {
uint32_t id; /* Only used by corosync derivatives */
uint64_t born; /* Only used by heartbeat and the legacy plugin */
uint64_t last_seen;
uint64_t flags; /* Specified by crm_node_flags enum */
int32_t votes; /* Only used by the legacy plugin */
uint32_t processes;
enum crm_join_phase join;
char *uname;
char *uuid;
char *state;
char *expected;
char *addr; /* Only used by the legacy plugin */
char *version; /* Unused */
} crm_node_t;
void crm_peer_init(void);
void crm_peer_destroy(void);
typedef struct crm_cluster_s {
char *uuid;
char *uname;
uint32_t nodeid;
void (*destroy) (gpointer);
# if SUPPORT_HEARTBEAT
ll_cluster_t *hb_conn;
void (*hb_dispatch) (HA_Message * msg, void *private);
# endif
# if SUPPORT_COROSYNC
struct cpg_name group;
cpg_callbacks_t cpg;
cpg_handle_t cpg_handle;
# endif
} crm_cluster_t;
gboolean crm_cluster_connect(crm_cluster_t * cluster);
void crm_cluster_disconnect(crm_cluster_t * cluster);
/* *INDENT-OFF* */
enum crm_ais_msg_class {
crm_class_cluster = 0,
crm_class_members = 1,
crm_class_notify = 2,
crm_class_nodeid = 3,
crm_class_rmpeer = 4,
crm_class_quorum = 5,
};
/* 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,
crm_msg_stonith_ng = 9,
};
/* used with crm_get_peer_full */
enum crm_get_peer_flags {
CRM_GET_PEER_CLUSTER = 0x0001,
CRM_GET_PEER_REMOTE = 0x0002,
};
/* *INDENT-ON* */
gboolean send_cluster_message(crm_node_t * node, enum crm_ais_msg_types service,
xmlNode * data, gboolean ordered);
int crm_remote_peer_cache_size(void);
/* Initialize and refresh the remote peer cache from a cib config */
void crm_remote_peer_cache_refresh(xmlNode *cib);
void crm_remote_peer_cache_add(const char *node_name);
void crm_remote_peer_cache_remove(const char *node_name);
/* allows filtering of remote and cluster nodes using crm_get_peer_flags */
crm_node_t *crm_get_peer_full(unsigned int id, const char *uname, int flags);
/* only searches cluster nodes */
crm_node_t *crm_get_peer(unsigned int id, const char *uname);
guint crm_active_peers(void);
gboolean crm_is_peer_active(const crm_node_t * node);
guint reap_crm_member(uint32_t id, const char *name);
int crm_terminate_member(int nodeid, const char *uname, void *unused);
int crm_terminate_member_no_mainloop(int nodeid, const char *uname, int *connection);
# if SUPPORT_HEARTBEAT
gboolean crm_is_heartbeat_peer_active(const crm_node_t * node);
# endif
# if SUPPORT_COROSYNC
extern int ais_fd_sync;
uint32_t get_local_nodeid(cpg_handle_t handle);
gboolean cluster_connect_cpg(crm_cluster_t *cluster);
void cluster_disconnect_cpg(crm_cluster_t * cluster);
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);
gboolean crm_is_corosync_peer_active(const crm_node_t * node);
gboolean send_cluster_text(int class, const char *data, gboolean local,
crm_node_t * node, enum crm_ais_msg_types dest);
# endif
const char *crm_peer_uuid(crm_node_t *node);
const char *crm_peer_uname(const char *uuid);
void set_uuid(xmlNode *xml, const char *attr, crm_node_t *node);
enum crm_status_type {
crm_status_uname,
crm_status_nstate,
crm_status_processes,
crm_status_rstate, /* remote node state */
};
enum crm_ais_msg_types text2msg_type(const char *text);
void crm_set_status_callback(void (*dispatch) (enum crm_status_type, crm_node_t *, const void *));
+void crm_set_autoreap(gboolean autoreap);
/* *INDENT-OFF* */
enum cluster_type_e
{
pcmk_cluster_unknown = 0x0001,
pcmk_cluster_invalid = 0x0002,
pcmk_cluster_heartbeat = 0x0004,
pcmk_cluster_classic_ais = 0x0010,
pcmk_cluster_corosync = 0x0020,
pcmk_cluster_cman = 0x0040,
};
/* *INDENT-ON* */
enum cluster_type_e get_cluster_type(void);
const char *name_for_cluster_type(enum cluster_type_e type);
gboolean is_corosync_cluster(void);
gboolean is_cman_cluster(void);
gboolean is_openais_cluster(void);
gboolean is_classic_ais_cluster(void);
gboolean is_heartbeat_cluster(void);
const char *get_local_node_name(void);
char *get_node_name(uint32_t nodeid);
# if SUPPORT_COROSYNC
char *pcmk_message_common_cs(cpg_handle_t handle, uint32_t nodeid, uint32_t pid, void *msg,
uint32_t *kind, const char **from);
# endif
#endif
diff --git a/include/crm/cluster/internal.h b/include/crm/cluster/internal.h
index bf15f96376..f915ae3d2a 100644
--- a/include/crm/cluster/internal.h
+++ b/include/crm/cluster/internal.h
@@ -1,476 +1,476 @@
/*
* Copyright (C) 2004 Andrew Beekhof <andrew@beekhof.net>
*
* 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 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef CRM_CLUSTER_INTERNAL__H
# define CRM_CLUSTER_INTERNAL__H
# include <crm/cluster.h>
# define AIS_IPC_NAME "ais-crm-ipc"
# define AIS_IPC_MESSAGE_SIZE 8192*128
# define CRM_MESSAGE_IPC_ACK 0
# ifndef INTERFACE_MAX
# define INTERFACE_MAX 2 /* from the private coroapi.h header */
# endif
typedef struct crm_ais_host_s AIS_Host;
typedef struct crm_ais_msg_s AIS_Message;
struct crm_ais_host_s {
uint32_t id;
uint32_t pid;
gboolean local;
enum crm_ais_msg_types type;
uint32_t size;
char uname[MAX_NAME];
} __attribute__ ((packed));
struct crm_ais_msg_s {
cs_ipc_header_response_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));
struct crm_ais_nodeid_resp_s {
cs_ipc_header_response_t header __attribute__ ((aligned(8)));
uint32_t id;
uint32_t counter;
char uname[MAX_NAME];
char cname[MAX_NAME];
} __attribute__ ((packed));
struct crm_ais_quorum_resp_s {
cs_ipc_header_response_t header __attribute__ ((aligned(8)));
uint64_t id;
uint32_t votes;
uint32_t expected_votes;
uint32_t quorate;
} __attribute__ ((packed));
/* *INDENT-OFF* */
enum crm_proc_flag {
crm_proc_none = 0x00000001,
/* These values are sent over the network by the legacy plugin
* Therefor changing any of these values is going to break compatability
*
* So don't
*/
/* 3 messaging types */
crm_proc_heartbeat = 0x01000000,
crm_proc_plugin = 0x00000002,
crm_proc_cpg = 0x04000000,
crm_proc_lrmd = 0x00000010,
crm_proc_cib = 0x00000100,
crm_proc_crmd = 0x00000200,
crm_proc_attrd = 0x00001000,
crm_proc_stonithd = 0x00002000,
crm_proc_stonith_ng= 0x00100000,
crm_proc_pe = 0x00010000,
crm_proc_te = 0x00020000,
crm_proc_mgmtd = 0x00040000,
};
/* *INDENT-ON* */
/*!
* \internal
* \brief Return the process bit corresponding to the current cluster stack
*
* \return Process flag if detectable, otherwise 0
*/
static inline uint32_t
crm_get_cluster_proc()
{
switch (get_cluster_type()) {
case pcmk_cluster_corosync:
case pcmk_cluster_cman:
return crm_proc_cpg;
case pcmk_cluster_heartbeat:
return crm_proc_heartbeat;
case pcmk_cluster_classic_ais:
return crm_proc_plugin;
default:
break;
}
return crm_proc_none;
}
static inline const char *
peer2text(enum crm_proc_flag proc)
{
const char *text = "unknown";
if (proc == (crm_proc_crmd | crm_get_cluster_proc())) {
return "peer";
}
switch (proc) {
case crm_proc_none:
text = "none";
break;
case crm_proc_plugin:
text = "ais";
break;
case crm_proc_heartbeat:
text = "heartbeat";
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;
case crm_proc_stonith_ng:
text = "stonith-ng";
break;
case crm_proc_mgmtd:
text = "mgmtd";
break;
case crm_proc_cpg:
text = "corosync-cpg";
break;
}
return text;
}
static inline enum crm_proc_flag
text2proc(const char *proc)
{
/* We only care about these two so far */
if (proc && strcmp(proc, "cib") == 0) {
return crm_proc_cib;
} else if (proc && strcmp(proc, "crmd") == 0) {
return crm_proc_crmd;
}
return crm_proc_none;
}
static inline const char *
ais_dest(const struct crm_ais_host_s *host)
{
if (host->local) {
return "local";
} else if (host->size > 0) {
return host->uname;
} else {
return "<all>";
}
}
# define ais_data_len(msg) (msg->is_compressed?msg->compressed_size:msg->size)
static inline AIS_Message *
ais_msg_copy(const AIS_Message * source)
{
AIS_Message *target = malloc(sizeof(AIS_Message) + ais_data_len(source));
if(target) {
memcpy(target, source, sizeof(AIS_Message));
memcpy(target->data, source->data, ais_data_len(target));
}
return target;
}
/*
typedef enum {
CS_OK = 1,
CS_ERR_LIBRARY = 2,
CS_ERR_VERSION = 3,
CS_ERR_INIT = 4,
CS_ERR_TIMEOUT = 5,
CS_ERR_TRY_AGAIN = 6,
CS_ERR_INVALID_PARAM = 7,
CS_ERR_NO_MEMORY = 8,
CS_ERR_BAD_HANDLE = 9,
CS_ERR_BUSY = 10,
CS_ERR_ACCESS = 11,
CS_ERR_NOT_EXIST = 12,
CS_ERR_NAME_TOO_LONG = 13,
CS_ERR_EXIST = 14,
CS_ERR_NO_SPACE = 15,
CS_ERR_INTERRUPT = 16,
CS_ERR_NAME_NOT_FOUND = 17,
CS_ERR_NO_RESOURCES = 18,
CS_ERR_NOT_SUPPORTED = 19,
CS_ERR_BAD_OPERATION = 20,
CS_ERR_FAILED_OPERATION = 21,
CS_ERR_MESSAGE_ERROR = 22,
CS_ERR_QUEUE_FULL = 23,
CS_ERR_QUEUE_NOT_AVAILABLE = 24,
CS_ERR_BAD_FLAGS = 25,
CS_ERR_TOO_BIG = 26,
CS_ERR_NO_SECTIONS = 27,
CS_ERR_CONTEXT_NOT_FOUND = 28,
CS_ERR_TOO_MANY_GROUPS = 30,
CS_ERR_SECURITY = 100
} cs_error_t;
*/
static inline const char *
ais_error2text(int error)
{
const char *text = "unknown";
# if SUPPORT_COROSYNC
switch (error) {
case CS_OK:
text = "OK";
break;
case CS_ERR_LIBRARY:
text = "Library error";
break;
case CS_ERR_VERSION:
text = "Version error";
break;
case CS_ERR_INIT:
text = "Initialization error";
break;
case CS_ERR_TIMEOUT:
text = "Timeout";
break;
case CS_ERR_TRY_AGAIN:
text = "Try again";
break;
case CS_ERR_INVALID_PARAM:
text = "Invalid parameter";
break;
case CS_ERR_NO_MEMORY:
text = "No memory";
break;
case CS_ERR_BAD_HANDLE:
text = "Bad handle";
break;
case CS_ERR_BUSY:
text = "Busy";
break;
case CS_ERR_ACCESS:
text = "Access error";
break;
case CS_ERR_NOT_EXIST:
text = "Doesn't exist";
break;
case CS_ERR_NAME_TOO_LONG:
text = "Name too long";
break;
case CS_ERR_EXIST:
text = "Exists";
break;
case CS_ERR_NO_SPACE:
text = "No space";
break;
case CS_ERR_INTERRUPT:
text = "Interrupt";
break;
case CS_ERR_NAME_NOT_FOUND:
text = "Name not found";
break;
case CS_ERR_NO_RESOURCES:
text = "No resources";
break;
case CS_ERR_NOT_SUPPORTED:
text = "Not supported";
break;
case CS_ERR_BAD_OPERATION:
text = "Bad operation";
break;
case CS_ERR_FAILED_OPERATION:
text = "Failed operation";
break;
case CS_ERR_MESSAGE_ERROR:
text = "Message error";
break;
case CS_ERR_QUEUE_FULL:
text = "Queue full";
break;
case CS_ERR_QUEUE_NOT_AVAILABLE:
text = "Queue not available";
break;
case CS_ERR_BAD_FLAGS:
text = "Bad flags";
break;
case CS_ERR_TOO_BIG:
text = "To big";
break;
case CS_ERR_NO_SECTIONS:
text = "No sections";
break;
}
# endif
return text;
}
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;
case crm_msg_stonith_ng:
text = "stonith-ng";
break;
}
return text;
}
enum crm_ais_msg_types text2msg_type(const char *text);
char *get_ais_data(const AIS_Message * msg);
gboolean check_message_sanity(const AIS_Message * msg, const char *data);
# if SUPPORT_HEARTBEAT
extern ll_cluster_t *heartbeat_cluster;
gboolean send_ha_message(ll_cluster_t * hb_conn, xmlNode * msg,
const char *node, gboolean force_ordered);
gboolean ha_msg_dispatch(ll_cluster_t * cluster_conn, gpointer user_data);
gboolean register_heartbeat_conn(crm_cluster_t * cluster);
xmlNode *convert_ha_message(xmlNode * parent, HA_Message * msg, const char *field);
gboolean ccm_have_quorum(oc_ed_t event);
const char *ccm_event_name(oc_ed_t event);
crm_node_t *crm_update_ccm_node(const oc_ev_membership_t * oc, int offset, const char *state,
uint64_t seq);
gboolean heartbeat_initialize_nodelist(void *cluster, gboolean force_member, xmlNode * xml_parent);
# endif
# if SUPPORT_COROSYNC
gboolean send_cpg_iov(struct iovec * iov);
# if SUPPORT_PLUGIN
char *classic_node_name(uint32_t nodeid);
void plugin_handle_membership(AIS_Message *msg);
bool send_plugin_text(int class, struct iovec *iov);
# else
char *corosync_node_name(uint64_t /*cmap_handle_t */ cmap_handle, uint32_t nodeid);
char *corosync_cluster_name(void);
int corosync_cmap_has_config(const char *prefix);
# endif
gboolean corosync_initialize_nodelist(void *cluster, gboolean force_member, xmlNode * xml_parent);
gboolean send_cluster_message_cs(xmlNode * msg, gboolean local,
crm_node_t * node, enum crm_ais_msg_types dest);
enum cluster_type_e find_corosync_variant(void);
void terminate_cs_connection(crm_cluster_t * cluster);
gboolean init_cs_connection(crm_cluster_t * cluster);
gboolean init_cs_connection_once(crm_cluster_t * cluster);
# endif
# ifdef SUPPORT_CMAN
char *cman_node_name(uint32_t nodeid);
# endif
enum crm_quorum_source {
crm_quorum_cman,
crm_quorum_corosync,
crm_quorum_pacemaker,
};
int get_corosync_id(int id, const char *uuid);
char *get_corosync_uuid(crm_node_t *peer);
enum crm_quorum_source get_quorum_source(void);
-void crm_update_peer_proc(const char *source, crm_node_t * peer, uint32_t flag, const char *status);
-
-crm_node_t *crm_update_peer(const char *source, unsigned int id, uint64_t born, uint64_t seen,
- int32_t votes, uint32_t children, const char *uuid, const char *uname,
+crm_node_t *crm_update_peer(const char *source, unsigned int id, uint64_t born,
+ uint64_t seen, int32_t votes, uint32_t children,
+ const char *uuid, const char *uname,
const char *addr, const char *state);
+crm_node_t *crm_update_peer_proc(const char *source, crm_node_t * peer,
+ uint32_t flag, const char *status);
+crm_node_t *crm_update_peer_state(const char *source, crm_node_t * node,
+ const char *state, int membership);
void crm_update_peer_expected(const char *source, crm_node_t * node, const char *expected);
-void crm_update_peer_state(const char *source, crm_node_t * node, const char *state,
- int membership);
void crm_reap_unseen_nodes(uint64_t ring_id);
-
gboolean init_cman_connection(gboolean(*dispatch) (unsigned long long, gboolean),
void (*destroy) (gpointer));
gboolean cluster_connect_quorum(gboolean(*dispatch) (unsigned long long, gboolean),
void (*destroy) (gpointer));
void set_node_uuid(const char *uname, const char *uuid);
gboolean node_name_is_valid(const char *key, const char *name);
crm_node_t * crm_find_peer_full(unsigned int id, const char *uname, int flags);
crm_node_t * crm_find_peer(unsigned int id, const char *uname);
#endif
diff --git a/lib/cluster/cpg.c b/lib/cluster/cpg.c
index 46c5fd4ee9..46edb7e966 100644
--- a/lib/cluster/cpg.c
+++ b/lib/cluster/cpg.c
@@ -1,686 +1,687 @@
/*
* 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->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_ONE);
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);
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_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_find_peer(left_list[i].nodeid, NULL);
crm_info("Node %u left group %s (peer=%s, counter=%d.%d)",
left_list[i].nodeid, groupName->value,
(peer? peer->uname : "<none>"), counter, i);
if (peer) {
crm_update_peer_proc(__FUNCTION__, peer, crm_proc_cpg, OFFLINESTATUS);
}
}
for (i = 0; i < joined_list_entries; i++) {
crm_info("Node %u joined group %s (counter=%d.%d)",
joined_list[i].nodeid, groupName->value, counter, i);
}
for (i = 0; i < member_list_entries; i++) {
crm_node_t *peer = crm_get_peer(member_list[i].nodeid, NULL);
crm_info("Node %u still member of group %s (peer=%s, counter=%d.%d)",
member_list[i].nodeid, groupName->value,
(peer? peer->uname : "<none>"), counter, i);
/* 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);
+ peer = 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 (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;
/* group.value is char[128] */
strncpy(cluster->group.value, crm_system_name, 127);
cluster->group.value[127] = 0;
cluster->group.length = 1 + QB_MIN(127, strlen(crm_system_name));
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 = crm_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);
if(local_name && msg->sender.size) {
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_safe(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;
msg = realloc_safe(msg, msg->header.size);
memcpy(msg->data, compressed, new_size);
msg->is_compressed = TRUE;
msg->compressed_size = new_size;
} else {
msg = realloc_safe(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);
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 || type <= crm_msg_stonith_ng) {
/* Ensure its sane */
type = crm_msg_none;
}
}
return type;
}
diff --git a/lib/cluster/heartbeat.c b/lib/cluster/heartbeat.c
index 3e8cfffd74..e55eeecd1b 100644
--- a/lib/cluster/heartbeat.c
+++ b/lib/cluster/heartbeat.c
@@ -1,657 +1,660 @@
/*
* 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 <dlfcn.h>
#include <sys/param.h>
#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <time.h>
#include <crm/crm.h>
#include <crm/msg_xml.h>
#include <crm/common/ipc.h>
#include <crm/cluster/internal.h>
#if HAVE_BZLIB_H
# include <bzlib.h>
#endif
#if SUPPORT_HEARTBEAT
ll_cluster_t *heartbeat_cluster = NULL;
static void
convert_ha_field(xmlNode * parent, void *msg_v, int lpc)
{
int type = 0;
const char *name = NULL;
const char *value = NULL;
xmlNode *xml = NULL;
HA_Message *msg = msg_v;
int rc = BZ_OK;
size_t orig_len = 0;
unsigned int used = 0;
char *uncompressed = NULL;
char *compressed = NULL;
int size = orig_len * 10;
CRM_CHECK(parent != NULL, return);
CRM_CHECK(msg != NULL, return);
name = msg->names[lpc];
type = cl_get_type(msg, name);
switch (type) {
case FT_STRUCT:
convert_ha_message(parent, msg->values[lpc], name);
break;
case FT_COMPRESS:
case FT_UNCOMPRESS:
convert_ha_message(parent, cl_get_struct(msg, name), name);
break;
case FT_STRING:
value = msg->values[lpc];
CRM_CHECK(value != NULL, return);
crm_trace("Converting %s/%d/%s", name, type, value[0] == '<' ? "xml" : "field");
if (value[0] != '<') {
crm_xml_add(parent, name, value);
break;
}
/* unpack xml string */
xml = string2xml(value);
if (xml == NULL) {
crm_err("Conversion of field '%s' failed", name);
return;
}
add_node_nocopy(parent, NULL, xml);
break;
case FT_BINARY:
value = cl_get_binary(msg, name, &orig_len);
size = orig_len * 10 + 1; /* +1 because an exact 10x compression factor happens occasionally */
if (orig_len < 3 || value[0] != 'B' || value[1] != 'Z' || value[2] != 'h') {
if (strstr(name, "uuid") == NULL) {
crm_err("Skipping non-bzip binary field: %s", name);
}
return;
}
compressed = calloc(1, orig_len);
memcpy(compressed, value, orig_len);
crm_trace("Trying to decompress %d bytes", (int)orig_len);
retry:
uncompressed = realloc_safe(uncompressed, size);
memset(uncompressed, 0, size);
used = size - 1; /* always leave room for a trailing '\0'
* BZ2_bzBuffToBuffDecompress wont say anything if
* the uncompressed data is exactly 'size' bytes
*/
rc = BZ2_bzBuffToBuffDecompress(uncompressed, &used, compressed, orig_len, 1, 0);
if (rc == BZ_OUTBUFF_FULL) {
size = size * 2;
/* dont try to allocate more memory than we have */
if (size > 0) {
goto retry;
}
}
if (rc != BZ_OK) {
crm_err("Decompression of %s (%d bytes) into %d failed: %d",
name, (int)orig_len, size, rc);
} else if (used >= size) {
CRM_ASSERT(used < size);
} else {
CRM_LOG_ASSERT(uncompressed[used] == 0);
uncompressed[used] = 0;
xml = string2xml(uncompressed);
}
if (xml != NULL) {
add_node_copy(parent, xml);
free_xml(xml);
}
free(uncompressed);
free(compressed);
break;
}
}
xmlNode *
convert_ha_message(xmlNode * parent, HA_Message * msg, const char *field)
{
int lpc = 0;
xmlNode *child = NULL;
const char *tag = NULL;
CRM_CHECK(msg != NULL, crm_err("Empty message for %s", field);
return parent);
tag = cl_get_string(msg, F_XML_TAGNAME);
if (tag == NULL) {
tag = field;
} else if (parent && safe_str_neq(field, tag)) {
/* For compatability with 0.6.x */
crm_debug("Creating intermediate parent %s between %s and %s", field,
crm_element_name(parent), tag);
parent = create_xml_node(parent, field);
}
if (parent == NULL) {
parent = create_xml_node(NULL, tag);
child = parent;
} else {
child = create_xml_node(parent, tag);
}
for (lpc = 0; lpc < msg->nfields; lpc++) {
convert_ha_field(child, msg, lpc);
}
return parent;
}
static void
add_ha_nocopy(HA_Message * parent, HA_Message * child, const char *field)
{
int next = parent->nfields;
if (parent->nfields >= parent->nalloc && ha_msg_expand(parent) != HA_OK) {
crm_err("Parent expansion failed");
return;
}
parent->names[next] = strdup(field);
parent->nlens[next] = strlen(field);
parent->values[next] = child;
parent->vlens[next] = sizeof(HA_Message);
parent->types[next] = FT_UNCOMPRESS;
parent->nfields++;
}
static HA_Message *
convert_xml_message_struct(HA_Message * parent, xmlNode * src_node, const char *field)
{
xmlNode *child = NULL;
xmlNode *__crm_xml_iter = src_node->children;
xmlAttrPtr prop_iter = src_node->properties;
const char *name = NULL;
const char *value = NULL;
HA_Message *result = ha_msg_new(3);
ha_msg_add(result, F_XML_TAGNAME, (const char *)src_node->name);
while (prop_iter != NULL) {
name = (const char *)prop_iter->name;
value = (const char *)xmlGetProp(src_node, prop_iter->name);
prop_iter = prop_iter->next;
ha_msg_add(result, name, value);
}
while (__crm_xml_iter != NULL) {
child = __crm_xml_iter;
__crm_xml_iter = __crm_xml_iter->next;
convert_xml_message_struct(result, child, NULL);
}
if (parent == NULL) {
return result;
}
if (field) {
HA_Message *holder = ha_msg_new(3);
CRM_ASSERT(holder != NULL);
ha_msg_add(holder, F_XML_TAGNAME, field);
add_ha_nocopy(holder, result, (const char *)src_node->name);
ha_msg_addstruct_compress(parent, field, holder);
ha_msg_del(holder);
} else {
add_ha_nocopy(parent, result, (const char *)src_node->name);
}
return result;
}
static void
convert_xml_child(HA_Message * msg, xmlNode * xml)
{
int orig = 0;
int rc = BZ_OK;
unsigned int len = 0;
char *buffer = NULL;
char *compressed = NULL;
const char *name = NULL;
name = (const char *)xml->name;
buffer = dump_xml_unformatted(xml);
orig = strlen(buffer);
if (orig < CRM_BZ2_THRESHOLD) {
ha_msg_add(msg, name, buffer);
goto done;
}
len = (orig * 1.1) + 600; /* recomended size */
compressed = malloc(len);
rc = BZ2_bzBuffToBuffCompress(compressed, &len, buffer, orig, CRM_BZ2_BLOCKS, 0, CRM_BZ2_WORK);
if (rc != BZ_OK) {
crm_err("Compression failed: %d", rc);
free(compressed);
convert_xml_message_struct(msg, xml, name);
goto done;
}
free(buffer);
buffer = compressed;
crm_trace("Compression details: %d -> %d", orig, len);
ha_msg_addbin(msg, name, buffer, len);
done:
free(buffer);
# if 0
{
unsigned int used = orig;
char *uncompressed = NULL;
crm_debug("Trying to decompress %d bytes", len);
uncompressed = calloc(1, orig);
rc = BZ2_bzBuffToBuffDecompress(uncompressed, &used, compressed, len, 1, 0);
CRM_CHECK(rc == BZ_OK,;
);
CRM_CHECK(used == orig,;
);
crm_debug("rc=%d, used=%d", rc, used);
if (rc != BZ_OK) {
crm_exit(DAEMON_RESPAWN_STOP);
}
crm_debug("Original %s, decompressed %s", buffer, uncompressed);
free(uncompressed);
}
# endif
}
static HA_Message *
convert_xml_message(xmlNode * xml)
{
xmlNode *child = NULL;
xmlAttrPtr pIter = NULL;
HA_Message *result = NULL;
result = ha_msg_new(3);
ha_msg_add(result, F_XML_TAGNAME, (const char *)xml->name);
for (pIter = xml->properties; pIter != NULL; pIter = pIter->next) {
const char *p_name = (const char *)pIter->name;
if (pIter->children) {
const char *p_value = (const char *)pIter->children->content;
ha_msg_add(result, p_name, p_value);
}
}
for (child = __xml_first_child(xml); child != NULL; child = __xml_next(child)) {
convert_xml_child(result, child);
}
return result;
}
gboolean
crm_is_heartbeat_peer_active(const crm_node_t * node)
{
enum crm_proc_flag proc = text2proc(crm_system_name);
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_heartbeat) == 0) {
crm_trace("%s: processes=%.16x", node->uname, node->processes);
return FALSE;
} else if (proc == crm_proc_none) {
return TRUE;
} else if ((node->processes & proc) == 0) {
crm_trace("%s: proc %.16x not in %.16x", node->uname, proc, node->processes);
return FALSE;
}
return TRUE;
}
crm_node_t *
crm_update_ccm_node(const oc_ev_membership_t * oc, int offset, const char *state, uint64_t seq)
{
enum crm_proc_flag this_proc = text2proc(crm_system_name);
crm_node_t *peer = NULL;
const char *uuid = NULL;
CRM_CHECK(oc->m_array[offset].node_uname != NULL, return NULL);
peer = crm_get_peer(0, oc->m_array[offset].node_uname);
uuid = crm_peer_uuid(peer);
- crm_update_peer(__FUNCTION__, oc->m_array[offset].node_id,
+ peer = crm_update_peer(__FUNCTION__, oc->m_array[offset].node_id,
oc->m_array[offset].node_born_on, seq, -1, 0,
uuid, oc->m_array[offset].node_uname, NULL, state);
+ if (peer == NULL) {
+ return NULL;
+ }
if (safe_str_eq(CRM_NODE_MEMBER, state)) {
/* Heartbeat doesn't send status notifications for nodes that were already part of the cluster.
* Nor does it send status notifications for processes that were already active.
* Do not optimistically assume the peer client process to be online as well.
* We ask for cluster wide updated client status for crm_system_name
* directly in the ccm status callback, which will then tell us.
* For ourselves, we know. */
enum crm_proc_flag flags = crm_proc_heartbeat;
const char *const_uname = heartbeat_cluster->llc_ops->get_mynodeid(heartbeat_cluster);
if (safe_str_eq(const_uname, peer->uname)) {
flags |= this_proc;
}
- crm_update_peer_proc(__FUNCTION__, peer, flags, ONLINESTATUS);
+ peer = crm_update_peer_proc(__FUNCTION__, peer, flags, ONLINESTATUS);
} else {
/* crm_update_peer_proc(__FUNCTION__, peer, crm_proc_heartbeat, OFFLINESTATUS); */
/* heartbeat may well be still alive. peer client process apparently vanished, though ... */
- crm_update_peer_proc(__FUNCTION__, peer, this_proc, OFFLINESTATUS);
+ peer = crm_update_peer_proc(__FUNCTION__, peer, this_proc, OFFLINESTATUS);
}
return peer;
}
gboolean
send_ha_message(ll_cluster_t * hb_conn, xmlNode * xml, const char *node, gboolean force_ordered)
{
gboolean all_is_good = TRUE;
HA_Message *msg = convert_xml_message(xml);
if (msg == NULL) {
crm_err("cant send NULL message");
all_is_good = FALSE;
} else if (hb_conn == NULL) {
crm_err("No heartbeat connection specified");
all_is_good = FALSE;
} else if (hb_conn->llc_ops->chan_is_connected(hb_conn) == FALSE) {
crm_err("Not connected to Heartbeat");
all_is_good = FALSE;
} else if (node != NULL) {
char *host_lowercase = g_ascii_strdown(node, -1);
if (hb_conn->llc_ops->send_ordered_nodemsg(hb_conn, msg, host_lowercase) != HA_OK) {
all_is_good = FALSE;
crm_err("Send failed");
}
free(host_lowercase);
} else if (force_ordered) {
if (hb_conn->llc_ops->send_ordered_clustermsg(hb_conn, msg) != HA_OK) {
all_is_good = FALSE;
crm_err("Broadcast Send failed");
}
} else {
if (hb_conn->llc_ops->sendclustermsg(hb_conn, msg) != HA_OK) {
all_is_good = FALSE;
crm_err("Broadcast Send failed");
}
}
if (all_is_good == FALSE && hb_conn != NULL) {
IPC_Channel *ipc = NULL;
IPC_Queue *send_q = NULL;
if (hb_conn->llc_ops->chan_is_connected(hb_conn) != HA_OK) {
ipc = hb_conn->llc_ops->ipcchan(hb_conn);
}
if (ipc != NULL) {
/* ipc->ops->resume_io(ipc); */
send_q = ipc->send_queue;
}
if (send_q != NULL) {
CRM_CHECK(send_q->current_qlen < send_q->max_qlen,;
);
}
}
if (all_is_good) {
crm_log_xml_trace(xml, "outbound");
} else {
crm_log_xml_warn(xml, "outbound");
}
if (msg != NULL) {
ha_msg_del(msg);
}
return all_is_good;
}
gboolean
ha_msg_dispatch(ll_cluster_t * cluster_conn, gpointer user_data)
{
IPC_Channel *channel = NULL;
crm_trace("Invoked");
if (cluster_conn != NULL) {
channel = cluster_conn->llc_ops->ipcchan(cluster_conn);
}
CRM_CHECK(cluster_conn != NULL, return FALSE);
CRM_CHECK(channel != NULL, return FALSE);
if (channel != NULL && IPC_ISRCONN(channel)) {
struct ha_msg *msg;
if (cluster_conn->llc_ops->msgready(cluster_conn) == 0) {
crm_trace("no message ready yet");
}
/* invoke the callbacks but dont block.
* cluster_conn->llc_ops->rcvmsg(cluster_conn, 0); */
msg = cluster_conn->llc_ops->readmsg(cluster_conn, 0);
if (msg) {
/* Message core refuses to pass on messages with F_TYPE not set.
* Messages with no specific F_TOID are notifications delivered to all.
*/
const char *msg_type = ha_msg_value(msg, F_TYPE) ?: "[type not set]";
const char *msg_to_id = ha_msg_value(msg, F_TOID);
if (safe_str_eq(msg_to_id, crm_system_name)) {
crm_err("Ignored incoming message. Please set_msg_callback on %s", msg_type);
} else if (msg_to_id) {
/* Message core will not deliver messages addressed to someone else to us.
* Are we not registered as crm_system_name? */
crm_notice("Ignored incoming message %s=%s %s=%s, please set_msg_callback",
F_TOID, msg_to_id, F_TYPE, msg_type);
} else {
crm_debug("Ignored incoming message %s=%s", F_TYPE, msg_type);
}
ha_msg_del(msg);
}
}
if (channel == NULL || channel->ch_status != IPC_CONNECT) {
crm_info("Lost connection to heartbeat service.");
return FALSE;
}
return TRUE;
}
gboolean
register_heartbeat_conn(crm_cluster_t * cluster)
{
crm_node_t *peer = NULL;
const char *const_uuid = NULL;
const char *const_uname = NULL;
crm_debug("Signing in with Heartbeat");
if (cluster->hb_conn->llc_ops->signon(cluster->hb_conn, crm_system_name) != HA_OK) {
crm_err("Cannot sign on with heartbeat: %s",
cluster->hb_conn->llc_ops->errmsg(cluster->hb_conn));
return FALSE;
}
if (HA_OK !=
cluster->hb_conn->llc_ops->set_msg_callback(cluster->hb_conn, crm_system_name,
cluster->hb_dispatch, cluster->hb_conn)) {
crm_err("Cannot set msg callback: %s", cluster->hb_conn->llc_ops->errmsg(cluster->hb_conn));
return FALSE;
} else {
void *handle = NULL;
GLLclusterSource *(*g_main_add_cluster) (int priority, ll_cluster_t * api,
gboolean can_recurse,
gboolean(*dispatch) (ll_cluster_t * source_data,
gpointer user_data),
gpointer userdata, GDestroyNotify notify) =
find_library_function(&handle, HEARTBEAT_LIBRARY, "G_main_add_ll_cluster", 1);
(*g_main_add_cluster) (G_PRIORITY_HIGH, cluster->hb_conn,
FALSE, ha_msg_dispatch, cluster->hb_conn, cluster->destroy);
dlclose(handle);
}
const_uname = cluster->hb_conn->llc_ops->get_mynodeid(cluster->hb_conn);
CRM_CHECK(const_uname != NULL, return FALSE);
peer = crm_get_peer(0, const_uname);
const_uuid = crm_peer_uuid(peer);
CRM_CHECK(const_uuid != NULL, return FALSE);
crm_info("Hostname: %s", const_uname);
crm_info("UUID: %s", const_uuid);
cluster->uname = strdup(const_uname);
cluster->uuid = strdup(const_uuid);
return TRUE;
}
gboolean
ccm_have_quorum(oc_ed_t event)
{
if (event == OC_EV_MS_NEW_MEMBERSHIP || event == OC_EV_MS_PRIMARY_RESTORED) {
return TRUE;
}
return FALSE;
}
const char *
ccm_event_name(oc_ed_t event)
{
if (event == OC_EV_MS_NEW_MEMBERSHIP) {
return "NEW MEMBERSHIP";
} else if (event == OC_EV_MS_NOT_PRIMARY) {
return "NOT PRIMARY";
} else if (event == OC_EV_MS_PRIMARY_RESTORED) {
return "PRIMARY RESTORED";
} else if (event == OC_EV_MS_EVICTED) {
return "EVICTED";
} else if (event == OC_EV_MS_INVALID) {
return "INVALID";
}
return "NO QUORUM MEMBERSHIP";
}
gboolean
heartbeat_initialize_nodelist(void *cluster, gboolean force_member, xmlNode * xml_parent)
{
const char *ha_node = NULL;
ll_cluster_t *conn = cluster;
if (conn == NULL) {
crm_debug("Not connected");
return FALSE;
}
/* Async get client status information in the cluster */
crm_info("Requesting the list of configured nodes");
conn->llc_ops->init_nodewalk(conn);
do {
xmlNode *node = NULL;
crm_node_t *peer = NULL;
const char *ha_node_type = NULL;
const char *ha_node_uuid = NULL;
ha_node = conn->llc_ops->nextnode(conn);
if (ha_node == NULL) {
continue;
}
ha_node_type = conn->llc_ops->node_type(conn, ha_node);
if (safe_str_neq(NORMALNODE, ha_node_type)) {
crm_debug("Node %s: skipping '%s'", ha_node, ha_node_type);
continue;
}
peer = crm_get_peer(0, ha_node);
ha_node_uuid = crm_peer_uuid(peer);
if (ha_node_uuid == NULL) {
crm_warn("Node %s: no uuid found", ha_node);
continue;
}
crm_debug("Node: %s (uuid: %s)", ha_node, ha_node_uuid);
node = create_xml_node(xml_parent, XML_CIB_TAG_NODE);
crm_xml_add(node, XML_ATTR_ID, ha_node_uuid);
crm_xml_add(node, XML_ATTR_UNAME, ha_node);
crm_xml_add(node, XML_ATTR_TYPE, ha_node_type);
} while (ha_node != NULL);
conn->llc_ops->end_nodewalk(conn);
return TRUE;
}
#endif
diff --git a/lib/cluster/membership.c b/lib/cluster/membership.c
index f632af79b7..28f41cb4b0 100644
--- a/lib/cluster/membership.c
+++ b/lib/cluster/membership.c
@@ -1,787 +1,887 @@
/*
* 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>
#ifndef _GNU_SOURCE
# define _GNU_SOURCE
#endif
#include <sys/param.h>
#include <sys/types.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <glib.h>
#include <crm/common/ipc.h>
#include <crm/cluster/internal.h>
#include <crm/msg_xml.h>
#include <crm/stonith-ng.h>
GHashTable *crm_peer_cache = NULL;
GHashTable *crm_remote_peer_cache = NULL;
unsigned long long crm_peer_seq = 0;
gboolean crm_have_quorum = FALSE;
+static gboolean crm_autoreap = TRUE;
int
crm_remote_peer_cache_size(void)
{
if (crm_remote_peer_cache == NULL) {
return 0;
}
return g_hash_table_size(crm_remote_peer_cache);
}
void
crm_remote_peer_cache_add(const char *node_name)
{
crm_node_t *node = g_hash_table_lookup(crm_remote_peer_cache, node_name);
if (node == NULL) {
crm_trace("added %s to remote cache", node_name);
node = calloc(1, sizeof(crm_node_t));
node->flags = crm_remote_node;
CRM_ASSERT(node);
node->uname = strdup(node_name);
node->uuid = strdup(node_name);
node->state = strdup(CRM_NODE_MEMBER);
g_hash_table_replace(crm_remote_peer_cache, node->uname, node);
}
}
void
crm_remote_peer_cache_remove(const char *node_name)
{
g_hash_table_remove(crm_remote_peer_cache, node_name);
}
static void
remote_cache_refresh_helper(xmlNode *cib, const char *xpath, const char *field, int flags)
{
const char *remote = NULL;
crm_node_t *node = NULL;
xmlXPathObjectPtr xpathObj = NULL;
int max = 0;
int lpc = 0;
xpathObj = xpath_search(cib, xpath);
max = numXpathResults(xpathObj);
for (lpc = 0; lpc < max; lpc++) {
xmlNode *xml = getXpathResult(xpathObj, lpc);
CRM_LOG_ASSERT(xml != NULL);
if(xml != NULL) {
remote = crm_element_value(xml, field);
}
if (remote) {
crm_trace("added %s to remote cache", remote);
node = calloc(1, sizeof(crm_node_t));
node->flags = flags;
CRM_ASSERT(node);
node->uname = strdup(remote);
node->uuid = strdup(remote);
node->state = strdup(CRM_NODE_MEMBER);
g_hash_table_replace(crm_remote_peer_cache, node->uname, node);
}
}
freeXpathObject(xpathObj);
}
void crm_remote_peer_cache_refresh(xmlNode *cib)
{
const char *xpath = NULL;
g_hash_table_remove_all(crm_remote_peer_cache);
/* remote nodes associated with a cluster resource */
xpath = "//" XML_TAG_CIB "//" XML_CIB_TAG_CONFIGURATION "//" XML_CIB_TAG_RESOURCE "//" XML_TAG_META_SETS "//" XML_CIB_TAG_NVPAIR "[@name='remote-node']";
remote_cache_refresh_helper(cib, xpath, "value", crm_remote_node | crm_remote_container);
/* baremetal nodes defined by connection resources*/
xpath = "//" XML_TAG_CIB "//" XML_CIB_TAG_CONFIGURATION "//" XML_CIB_TAG_RESOURCE "[@type='remote'][@provider='pacemaker']";
remote_cache_refresh_helper(cib, xpath, "id", crm_remote_node | crm_remote_baremetal);
/* baremetal nodes we have seen in the config that may or may not have connection
* resources associated with them anymore */
xpath = "//" XML_TAG_CIB "//" XML_CIB_TAG_STATUS "//" XML_CIB_TAG_STATE "[@remote_node='true']";
remote_cache_refresh_helper(cib, xpath, "id", crm_remote_node | crm_remote_baremetal);
}
gboolean
crm_is_peer_active(const crm_node_t * node)
{
if(node == NULL) {
return FALSE;
}
if (is_set(node->flags, crm_remote_node)) {
/* remote nodes are never considered active members. This
* guarantees they will never be considered for DC membership.*/
return FALSE;
}
#if SUPPORT_COROSYNC
if (is_openais_cluster()) {
return crm_is_corosync_peer_active(node);
}
#endif
#if SUPPORT_HEARTBEAT
if (is_heartbeat_cluster()) {
return crm_is_heartbeat_peer_active(node);
}
#endif
crm_err("Unhandled cluster type: %s", name_for_cluster_type(get_cluster_type()));
return FALSE;
}
static gboolean
crm_reap_dead_member(gpointer key, gpointer value, gpointer user_data)
{
crm_node_t *node = value;
crm_node_t *search = user_data;
if (search == NULL) {
return FALSE;
} else if (search->id && node->id != search->id) {
return FALSE;
} else if (search->id == 0 && safe_str_neq(node->uname, search->uname)) {
return FALSE;
} else if (crm_is_peer_active(value) == FALSE) {
crm_notice("Removing %s/%u from the membership list", node->uname, node->id);
return TRUE;
}
return FALSE;
}
+/*!
+ * \brief Remove all peer cache entries matching a node ID and/or uname
+ *
+ * \param[in] id ID of node to remove (or 0 to ignore)
+ * \param[in] name Uname of node to remove (or NULL to ignore)
+ *
+ * \return Number of cache entries removed
+ */
guint
reap_crm_member(uint32_t id, const char *name)
{
int matches = 0;
crm_node_t search;
if (crm_peer_cache == NULL) {
crm_trace("Nothing to do, cache not initialized");
return 0;
}
search.id = id;
search.uname = name ? strdup(name) : NULL;
matches = g_hash_table_foreach_remove(crm_peer_cache, crm_reap_dead_member, &search);
if(matches) {
crm_notice("Purged %d peers with id=%u and/or uname=%s from the membership cache",
matches, search.id, search.uname);
} else {
crm_info("No peers with id=%u and/or uname=%s exist", id, name);
}
free(search.uname);
return matches;
}
static void
crm_count_peer(gpointer key, gpointer value, gpointer user_data)
{
guint *count = user_data;
crm_node_t *node = value;
if (crm_is_peer_active(node)) {
*count = *count + 1;
}
}
guint
crm_active_peers(void)
{
guint count = 0;
if (crm_peer_cache) {
g_hash_table_foreach(crm_peer_cache, crm_count_peer, &count);
}
return count;
}
static void
destroy_crm_node(gpointer data)
{
crm_node_t *node = data;
crm_trace("Destroying entry for node %u: %s", node->id, node->uname);
free(node->addr);
free(node->uname);
free(node->state);
free(node->uuid);
free(node->expected);
free(node);
}
void
crm_peer_init(void)
{
if (crm_peer_cache == NULL) {
crm_peer_cache = g_hash_table_new_full(crm_strcase_hash, crm_strcase_equal, free, destroy_crm_node);
}
if (crm_remote_peer_cache == NULL) {
crm_remote_peer_cache = g_hash_table_new_full(crm_strcase_hash, crm_strcase_equal, NULL, destroy_crm_node);
}
}
void
crm_peer_destroy(void)
{
if (crm_peer_cache != NULL) {
crm_trace("Destroying peer cache with %d members", g_hash_table_size(crm_peer_cache));
g_hash_table_destroy(crm_peer_cache);
crm_peer_cache = NULL;
}
if (crm_remote_peer_cache != NULL) {
crm_trace("Destroying remote peer cache with %d members", g_hash_table_size(crm_remote_peer_cache));
g_hash_table_destroy(crm_remote_peer_cache);
crm_remote_peer_cache = NULL;
}
}
void (*crm_status_callback) (enum crm_status_type, crm_node_t *, const void *) = NULL;
+/*!
+ * \brief Set a client function that will be called after peer status changes
+ *
+ * \param[in] dispatch Pointer to function to use as callback
+ *
+ * \note Previously, client callbacks were responsible for peer cache
+ * management. This is no longer the case, and client callbacks should do
+ * only client-specific handling. Callbacks MUST NOT add or remove entries
+ * in the peer caches.
+ */
void
crm_set_status_callback(void (*dispatch) (enum crm_status_type, crm_node_t *, const void *))
{
crm_status_callback = dispatch;
}
+/*!
+ * \brief Tell the library whether to automatically reap lost nodes
+ *
+ * If TRUE (the default), calling crm_update_peer_proc() will also update the
+ * peer state to CRM_NODE_MEMBER or CRM_NODE_LOST, and crm_update_peer_state()
+ * will reap peers whose state changes to anything other than CRM_NODE_MEMBER.
+ * Callers should leave this enabled unless they plan to manage the cache
+ * separately on their own.
+ *
+ * \param[in] autoreap TRUE to enable automatic reaping, FALSE to disable
+ */
+void
+crm_set_autoreap(gboolean autoreap)
+{
+ crm_autoreap = autoreap;
+}
+
static void crm_dump_peer_hash(int level, const char *caller)
{
GHashTableIter iter;
const char *id = NULL;
crm_node_t *node = NULL;
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, (gpointer *) &id, (gpointer *) &node)) {
do_crm_log(level, "%s: Node %u/%s = %p - %s", caller, node->id, node->uname, node, id);
}
}
static gboolean crm_hash_find_by_data(gpointer key, gpointer value, gpointer user_data)
{
if(value == user_data) {
return TRUE;
}
return FALSE;
}
crm_node_t *
crm_find_peer_full(unsigned int id, const char *uname, int flags)
{
crm_node_t *node = NULL;
CRM_ASSERT(id > 0 || uname != NULL);
crm_peer_init();
if (flags & CRM_GET_PEER_REMOTE) {
node = g_hash_table_lookup(crm_remote_peer_cache, uname);
}
if (node == NULL && (flags & CRM_GET_PEER_CLUSTER)) {
node = crm_find_peer(id, uname);
}
return node;
}
crm_node_t *
crm_get_peer_full(unsigned int id, const char *uname, int flags)
{
crm_node_t *node = NULL;
CRM_ASSERT(id > 0 || uname != NULL);
crm_peer_init();
if (flags & CRM_GET_PEER_REMOTE) {
node = g_hash_table_lookup(crm_remote_peer_cache, uname);
}
if (node == NULL && (flags & CRM_GET_PEER_CLUSTER)) {
node = crm_get_peer(id, uname);
}
return node;
}
crm_node_t *
crm_find_peer(unsigned int id, const char *uname)
{
GHashTableIter iter;
crm_node_t *node = NULL;
crm_node_t *by_id = NULL;
crm_node_t *by_name = NULL;
CRM_ASSERT(id > 0 || uname != NULL);
crm_peer_init();
if (uname != NULL) {
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &node)) {
if(node->uname && strcasecmp(node->uname, uname) == 0) {
crm_trace("Name match: %s = %p", node->uname, node);
by_name = node;
break;
}
}
}
if (id > 0) {
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &node)) {
if(node->id == id) {
crm_trace("ID match: %u = %p", node->id, node);
by_id = node;
break;
}
}
}
node = by_id; /* Good default */
if(by_id == by_name) {
/* Nothing to do if they match (both NULL counts) */
crm_trace("Consistent: %p for %u/%s", by_id, id, uname);
} else if(by_id == NULL && by_name) {
crm_trace("Only one: %p for %u/%s", by_name, id, uname);
if(id && by_name->id) {
crm_dump_peer_hash(LOG_WARNING, __FUNCTION__);
crm_crit("Node %u and %u share the same name '%s'",
id, by_name->id, uname);
node = NULL; /* Create a new one */
} else {
node = by_name;
}
} else if(by_name == NULL && by_id) {
crm_trace("Only one: %p for %u/%s", by_id, id, uname);
if(uname && by_id->uname) {
crm_dump_peer_hash(LOG_WARNING, __FUNCTION__);
crm_crit("Node '%s' and '%s' share the same cluster nodeid %u: assuming '%s' is correct",
uname, by_id->uname, id, uname);
}
} else if(uname && by_id->uname) {
if(safe_str_eq(uname, by_id->uname)) {
crm_notice("Node '%s' has changed its ID from %u to %u", by_id->uname, by_name->id, by_id->id);
g_hash_table_foreach_remove(crm_peer_cache, crm_hash_find_by_data, by_name);
} else {
crm_warn("Node '%s' and '%s' share the same cluster nodeid: %u %s", by_id->uname, by_name->uname, id, uname);
crm_dump_peer_hash(LOG_INFO, __FUNCTION__);
crm_abort(__FILE__, __FUNCTION__, __LINE__, "member weirdness", TRUE, TRUE);
}
} else if(id && by_name->id) {
crm_warn("Node %u and %u share the same name: '%s'", by_id->id, by_name->id, uname);
} else {
/* Simple merge */
/* Only corosync based clusters use nodeid's
*
* The functions that call crm_update_peer_state() only know nodeid
* so 'by_id' is authorative when merging
*
* Same for crm_update_peer_proc()
*/
crm_dump_peer_hash(LOG_DEBUG, __FUNCTION__);
crm_info("Merging %p into %p", by_name, by_id);
g_hash_table_foreach_remove(crm_peer_cache, crm_hash_find_by_data, by_name);
}
return node;
}
#if SUPPORT_COROSYNC
static guint
crm_remove_conflicting_peer(crm_node_t *node)
{
int matches = 0;
GHashTableIter iter;
crm_node_t *existing_node = NULL;
if (node->id == 0 || node->uname == NULL) {
return 0;
}
# if !SUPPORT_PLUGIN
if (corosync_cmap_has_config("nodelist") != 0) {
return 0;
}
# endif
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &existing_node)) {
if (existing_node->id > 0
&& existing_node->id != node->id
&& existing_node->uname != NULL
&& strcasecmp(existing_node->uname, node->uname) == 0) {
if (crm_is_peer_active(existing_node)) {
continue;
}
crm_warn("Removing cached offline node %u/%s which has conflicting uname with %u",
existing_node->id, existing_node->uname, node->id);
g_hash_table_iter_remove(&iter);
matches++;
}
}
return matches;
}
#endif
/* coverity[-alloc] Memory is referenced in one or both hashtables */
crm_node_t *
crm_get_peer(unsigned int id, const char *uname)
{
crm_node_t *node = NULL;
char *uname_lookup = NULL;
CRM_ASSERT(id > 0 || uname != NULL);
crm_peer_init();
node = crm_find_peer(id, uname);
/* if uname wasn't provided, and find_peer did not turn up a uname based on id.
* we need to do a lookup of the node name using the id in the cluster membership. */
if ((node == NULL || node->uname == NULL) && (uname == NULL)) {
uname_lookup = get_node_name(id);
}
if (uname_lookup) {
uname = uname_lookup;
crm_trace("Inferred a name of '%s' for node %u", uname, id);
/* try to turn up the node one more time now that we know the uname. */
if (node == NULL) {
node = crm_find_peer(id, uname);
}
}
if (node == NULL) {
char *uniqueid = crm_generate_uuid();
node = calloc(1, sizeof(crm_node_t));
CRM_ASSERT(node);
crm_info("Created entry %s/%p for node %s/%u (%d total)",
uniqueid, node, uname, id, 1 + g_hash_table_size(crm_peer_cache));
g_hash_table_replace(crm_peer_cache, uniqueid, node);
}
if(id > 0 && uname && (node->id == 0 || node->uname == NULL)) {
crm_info("Node %u is now known as %s", id, uname);
}
if(id > 0 && node->id == 0) {
node->id = id;
}
if(uname && node->uname == NULL) {
int lpc, len = strlen(uname);
for (lpc = 0; lpc < len; lpc++) {
if (uname[lpc] >= 'A' && uname[lpc] <= 'Z') {
crm_warn("Node names with capitals are discouraged, consider changing '%s' to something else",
uname);
break;
}
}
node->uname = strdup(uname);
if (crm_status_callback) {
crm_status_callback(crm_status_uname, node, NULL);
}
#if SUPPORT_COROSYNC
if (is_openais_cluster()) {
crm_remove_conflicting_peer(node);
}
#endif
}
if(node->uuid == NULL) {
const char *uuid = crm_peer_uuid(node);
if (uuid) {
crm_info("Node %u has uuid %s", id, uuid);
} else {
crm_info("Cannot obtain a UUID for node %u/%s", id, node->uname);
}
}
free(uname_lookup);
return node;
}
+/*!
+ * \internal
+ * \brief Update all of a node's information (process list, state, etc.)
+ *
+ * \param[in] source Caller's function name (for log messages)
+ *
+ * \return NULL if node was reaped from peer caches, pointer to node otherwise
+ *
+ * \note This function should not be called within a peer cache iteration,
+ * otherwise reaping could invalidate the iterator.
+ */
crm_node_t *
crm_update_peer(const char *source, unsigned int id, uint64_t born, uint64_t seen, int32_t votes,
uint32_t children, const char *uuid, const char *uname, const char *addr,
const char *state)
{
#if SUPPORT_PLUGIN
gboolean addr_changed = FALSE;
gboolean votes_changed = FALSE;
#endif
crm_node_t *node = NULL;
id = get_corosync_id(id, uuid);
node = crm_get_peer(id, uname);
CRM_ASSERT(node != NULL);
if (node->uuid == NULL) {
if (is_openais_cluster()) {
/* Yes, overrule whatever was passed in */
crm_peer_uuid(node);
} else if (uuid != NULL) {
node->uuid = strdup(uuid);
}
}
if (children > 0) {
- crm_update_peer_proc(source, node, children, state);
+ if (crm_update_peer_proc(source, node, children, state) == NULL) {
+ return NULL;
+ }
}
if (state != NULL) {
- crm_update_peer_state(source, node, state, seen);
+ if (crm_update_peer_state(source, node, state, seen) == NULL) {
+ return NULL;
+ }
}
#if SUPPORT_HEARTBEAT
if (born != 0) {
node->born = born;
}
#endif
#if SUPPORT_PLUGIN
/* These were only used by the plugin */
if (born != 0) {
node->born = born;
}
if (votes > 0 && node->votes != votes) {
votes_changed = TRUE;
node->votes = votes;
}
if (addr != NULL) {
if (node->addr == NULL || crm_str_eq(node->addr, addr, FALSE) == FALSE) {
addr_changed = TRUE;
free(node->addr);
node->addr = strdup(addr);
}
}
if (addr_changed || votes_changed) {
crm_info("%s: Node %s: id=%u state=%s addr=%s%s votes=%d%s born=" U64T " seen=" U64T
" proc=%.32x", source, node->uname, node->id, node->state,
node->addr, addr_changed ? " (new)" : "", node->votes,
votes_changed ? " (new)" : "", node->born, node->last_seen, node->processes);
}
#endif
return node;
}
-void
+/*!
+ * \internal
+ * \brief Update a node's process information (and potentially state)
+ *
+ * \param[in] source Caller's function name (for log messages)
+ * \param[in] node Node object to update
+ * \param[in] flag Bitmask of new process information
+ * \param[in] status node status (online, offline, etc.)
+ *
+ * \return NULL if any node was reaped from peer caches, value of node otherwise
+ *
+ * \note If this function returns TRUE, the supplied node object was likely
+ * freed and should not be used again. This function should not be
+ * called within a cache iteration if reaping is possible, otherwise
+ * reaping could invalidate the iterator.
+ */
+crm_node_t *
crm_update_peer_proc(const char *source, crm_node_t * node, uint32_t flag, const char *status)
{
uint32_t last = 0;
gboolean changed = FALSE;
CRM_CHECK(node != NULL, crm_err("%s: Could not set %s to %s for NULL",
- source, peer2text(flag), status); return);
+ source, peer2text(flag), status); return NULL);
last = node->processes;
if (status == NULL) {
node->processes = flag;
if (node->processes != last) {
changed = TRUE;
}
} else if (safe_str_eq(status, ONLINESTATUS)) {
if ((node->processes & flag) != flag) {
set_bit(node->processes, flag);
changed = TRUE;
}
#if SUPPORT_PLUGIN
} else if (safe_str_eq(status, CRM_NODE_MEMBER)) {
if (flag > 0 && node->processes != flag) {
node->processes = flag;
changed = TRUE;
}
#endif
} else if (node->processes & flag) {
clear_bit(node->processes, flag);
changed = TRUE;
}
if (changed) {
if (status == NULL && flag <= crm_proc_none) {
crm_info("%s: Node %s[%u] - all processes are now offline", source, node->uname,
node->id);
} else {
crm_info("%s: Node %s[%u] - %s is now %s", source, node->uname, node->id,
peer2text(flag), status);
}
+ /* Call the client callback first, then update the peer state,
+ * in case the node will be reaped
+ */
if (crm_status_callback) {
crm_status_callback(crm_status_processes, node, &last);
}
+ if (crm_autoreap) {
+ node = crm_update_peer_state(__FUNCTION__, node,
+ is_set(node->processes, crm_get_cluster_proc())?
+ CRM_NODE_MEMBER : CRM_NODE_LOST, 0);
+ }
} else {
crm_trace("%s: Node %s[%u] - %s is unchanged (%s)", source, node->uname, node->id,
peer2text(flag), status);
}
+ return node;
}
void
crm_update_peer_expected(const char *source, crm_node_t * node, const char *expected)
{
char *last = NULL;
gboolean changed = FALSE;
CRM_CHECK(node != NULL, crm_err("%s: Could not set 'expected' to %s", source, expected);
return);
last = node->expected;
if (expected != NULL && safe_str_neq(node->expected, expected)) {
node->expected = strdup(expected);
changed = TRUE;
}
if (changed) {
crm_info("%s: Node %s[%u] - expected state is now %s (was %s)", source, node->uname, node->id,
expected, last);
free(last);
} else {
crm_trace("%s: Node %s[%u] - expected state is unchanged (%s)", source, node->uname,
node->id, expected);
}
}
-void
+/*!
+ * \internal
+ * \brief Update a node's state and membership information
+ *
+ * \param[in] source Caller's function name (for log messages)
+ * \param[in] node Node object to update
+ * \param[in] state Node's new state
+ * \param[in] membership Node's new membership ID
+ *
+ * \return NULL if any node was reaped, value of node otherwise
+ *
+ * \note If this function returns NULL, the supplied node object was likely
+ * freed and should not be used again. This function should not be
+ * called within a cache iteration if reaping is possible,
+ * otherwise reaping could invalidate the iterator.
+ */
+crm_node_t *
crm_update_peer_state(const char *source, crm_node_t * node, const char *state, int membership)
{
- char *last = NULL;
- gboolean changed = FALSE;
+ gboolean is_member;
CRM_CHECK(node != NULL, crm_err("%s: Could not set 'state' to %s", source, state);
- return);
-
- last = node->state;
- if (state != NULL && safe_str_neq(node->state, state)) {
- node->state = strdup(state);
- changed = TRUE;
- }
+ return NULL);
- if (membership != 0 && safe_str_eq(node->state, CRM_NODE_MEMBER)) {
+ is_member = safe_str_eq(state, CRM_NODE_MEMBER);
+ if (membership && is_member) {
node->last_seen = membership;
}
- if (changed) {
- crm_notice("%s: Node %s[%u] - state is now %s (was %s)", source, node->uname, node->id, state, last);
+ if (state && safe_str_neq(node->state, state)) {
+ char *last = node->state;
+ enum crm_status_type status_type = is_set(node->flags, crm_remote_node)?
+ crm_status_rstate : crm_status_nstate;
+
+ node->state = strdup(state);
+ crm_notice("%s: Node %s[%u] - state is now %s (was %s)",
+ source, node->uname, node->id, state, last);
if (crm_status_callback) {
- enum crm_status_type status_type = crm_status_nstate;
- if (is_set(node->flags, crm_remote_node)) {
- status_type = crm_status_rstate;
- }
crm_status_callback(status_type, node, last);
}
free(last);
+
+ if (!is_member && crm_autoreap) {
+ if (status_type == crm_status_rstate) {
+ crm_remote_peer_cache_remove(node->uname);
+ } else {
+ reap_crm_member(node->id, node->uname);
+ }
+ node = NULL;
+ }
} else {
crm_trace("%s: Node %s[%u] - state is unchanged (%s)", source, node->uname, node->id,
state);
}
+ return node;
}
/*!
* \internal
* \brief Reap all nodes from cache whose membership information does not match
*
* \param[in] membership Membership ID of nodes to keep
*/
void
crm_reap_unseen_nodes(uint64_t membership)
{
GHashTableIter iter;
crm_node_t *node = NULL;
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 != membership) {
if (node->state) {
/* crm_update_peer_state() cannot be called here, because that
* might modify the peer cache, invalidating our iterator
*/
if (safe_str_eq(node->state, CRM_NODE_LOST)) {
crm_trace("Node %s[%u] - state is unchanged (%s)",
node->uname, node->id, CRM_NODE_LOST);
} else {
char *last = node->state;
node->state = strdup(CRM_NODE_LOST);
crm_notice("Node %s[%u] - state is now %s (was %s)",
node->uname, node->id, CRM_NODE_LOST, last);
if (crm_status_callback) {
crm_status_callback(crm_status_nstate, node, last);
}
+ if (crm_autoreap) {
+ g_hash_table_iter_remove(&iter);
+ }
free(last);
- g_hash_table_iter_remove(&iter);
}
} else {
crm_info("State of node %s[%u] is still unknown",
node->uname, node->id);
}
}
}
}
int
crm_terminate_member(int nodeid, const char *uname, void *unused)
{
/* Always use the synchronous, non-mainloop version */
return stonith_api_kick(nodeid, uname, 120, TRUE);
}
int
crm_terminate_member_no_mainloop(int nodeid, const char *uname, int *connection)
{
return stonith_api_kick(nodeid, uname, 120, TRUE);
}
diff --git a/mcp/pacemaker.c b/mcp/pacemaker.c
index f885e9ae58..1eb3144393 100644
--- a/mcp/pacemaker.c
+++ b/mcp/pacemaker.c
@@ -1,1139 +1,1144 @@
/*
* Copyright (C) 2010 Andrew Beekhof <andrew@beekhof.net>
*
* 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 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <crm_internal.h>
#include <pacemaker.h>
#include <pwd.h>
#include <grp.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <sys/reboot.h>
#include <crm/msg_xml.h>
#include <crm/common/ipcs.h>
#include <crm/common/mainloop.h>
#include <crm/cluster/internal.h>
#include <crm/cluster.h>
#include <dirent.h>
#include <ctype.h>
gboolean fatal_error = FALSE;
GMainLoop *mainloop = NULL;
#define PCMK_PROCESS_CHECK_INTERVAL 5
const char *local_name = NULL;
uint32_t local_nodeid = 0;
crm_trigger_t *shutdown_trigger = NULL;
const char *pid_file = "/var/run/pacemaker.pid";
typedef struct pcmk_child_s {
int pid;
long flag;
int start_seq;
int respawn_count;
gboolean respawn;
const char *name;
const char *uid;
const char *command;
gboolean active_before_startup;
} pcmk_child_t;
/* Index into the array below */
#define pcmk_child_crmd 4
#define pcmk_child_mgmtd 8
/* *INDENT-OFF* */
static pcmk_child_t pcmk_children[] = {
{ 0, crm_proc_none, 0, 0, FALSE, "none", NULL, NULL },
{ 0, crm_proc_plugin, 0, 0, FALSE, "ais", NULL, NULL },
{ 0, crm_proc_lrmd, 3, 0, TRUE, "lrmd", NULL, CRM_DAEMON_DIR"/lrmd" },
{ 0, crm_proc_cib, 1, 0, TRUE, "cib", CRM_DAEMON_USER, CRM_DAEMON_DIR"/cib" },
{ 0, crm_proc_crmd, 6, 0, TRUE, "crmd", CRM_DAEMON_USER, CRM_DAEMON_DIR"/crmd" },
{ 0, crm_proc_attrd, 4, 0, TRUE, "attrd", CRM_DAEMON_USER, CRM_DAEMON_DIR"/attrd" },
{ 0, crm_proc_stonithd, 0, 0, TRUE, "stonithd", NULL, NULL },
{ 0, crm_proc_pe, 5, 0, TRUE, "pengine", CRM_DAEMON_USER, CRM_DAEMON_DIR"/pengine" },
{ 0, crm_proc_mgmtd, 0, 0, TRUE, "mgmtd", NULL, HB_DAEMON_DIR"/mgmtd" },
{ 0, crm_proc_stonith_ng, 2, 0, TRUE, "stonith-ng", NULL, CRM_DAEMON_DIR"/stonithd" },
};
/* *INDENT-ON* */
static gboolean start_child(pcmk_child_t * child);
static gboolean check_active_before_startup_processes(gpointer user_data);
void update_process_clients(crm_client_t *client);
void update_process_peers(void);
void
enable_crmd_as_root(gboolean enable)
{
if (enable) {
pcmk_children[pcmk_child_crmd].uid = NULL;
} else {
pcmk_children[pcmk_child_crmd].uid = CRM_DAEMON_USER;
}
}
void
enable_mgmtd(gboolean enable)
{
if (enable) {
pcmk_children[pcmk_child_mgmtd].start_seq = 7;
} else {
pcmk_children[pcmk_child_mgmtd].start_seq = 0;
}
}
static uint32_t
get_process_list(void)
{
int lpc = 0;
uint32_t procs = crm_get_cluster_proc();
for (lpc = 0; lpc < SIZEOF(pcmk_children); lpc++) {
if (pcmk_children[lpc].pid != 0) {
procs |= pcmk_children[lpc].flag;
}
}
return procs;
}
static void
pcmk_process_exit(pcmk_child_t * child)
{
child->pid = 0;
child->active_before_startup = FALSE;
/* Broadcast the fact that one of our processes died ASAP
*
* Try to get some logging of the cause out first though
* because we're probably about to get fenced
*
* Potentially do this only if respawn_count > N
* to allow for local recovery
*/
update_node_processes(local_nodeid, NULL, get_process_list());
child->respawn_count += 1;
if (child->respawn_count > MAX_RESPAWN) {
crm_err("Child respawn count exceeded by %s", child->name);
child->respawn = FALSE;
}
if (shutdown_trigger) {
mainloop_set_trigger(shutdown_trigger);
update_node_processes(local_nodeid, NULL, get_process_list());
} else if (child->respawn && crm_is_true(getenv("PCMK_fail_fast"))) {
crm_err("Rebooting system because of %s", child->name);
pcmk_panic(__FUNCTION__);
} else if (child->respawn) {
crm_notice("Respawning failed child process: %s", child->name);
start_child(child);
}
}
static void
pcmk_child_exit(mainloop_child_t * p, pid_t pid, int core, int signo, int exitcode)
{
pcmk_child_t *child = mainloop_child_userdata(p);
const char *name = mainloop_child_name(p);
if (signo && signo == SIGKILL) {
crm_warn("The %s process (%d) terminated with signal %d (core=%d)", name, pid, signo, core);
} else if (signo) {
crm_err("The %s process (%d) terminated with signal %d (core=%d)", name, pid, signo, core);
} else {
switch(exitcode) {
case pcmk_ok:
crm_info("The %s process (%d) exited: %s (%d)", name, pid, pcmk_strerror(exitcode), exitcode);
break;
case DAEMON_RESPAWN_STOP:
crm_warn("The %s process (%d) can no longer be respawned, shutting the cluster down.", name, pid);
child->respawn = FALSE;
fatal_error = TRUE;
pcmk_shutdown(SIGTERM);
break;
case pcmk_err_panic:
do_crm_log_always(LOG_EMERG, "The %s process (%d) instructed the machine to reset", name, pid);
child->respawn = FALSE;
fatal_error = TRUE;
pcmk_panic(__FUNCTION__);
pcmk_shutdown(SIGTERM);
break;
default:
crm_err("The %s process (%d) exited: %s (%d)", name, pid, pcmk_strerror(exitcode), exitcode);
break;
}
}
pcmk_process_exit(child);
}
static gboolean
stop_child(pcmk_child_t * child, int signal)
{
if (signal == 0) {
signal = SIGTERM;
}
if (child->command == NULL) {
crm_debug("Nothing to do for child \"%s\"", child->name);
return TRUE;
}
if (child->pid <= 0) {
crm_trace("Client %s not running", child->name);
return TRUE;
}
errno = 0;
if (kill(child->pid, signal) == 0) {
crm_notice("Stopping %s: Sent -%d to process %d", child->name, signal, child->pid);
} else {
crm_perror(LOG_ERR, "Stopping %s: Could not send -%d to process %d failed",
child->name, signal, child->pid);
}
return TRUE;
}
static char *opts_default[] = { NULL, NULL };
static char *opts_vgrind[] = { NULL, NULL, NULL, NULL, NULL };
static gboolean
start_child(pcmk_child_t * child)
{
int lpc = 0;
uid_t uid = 0;
gid_t gid = 0;
struct rlimit oflimits;
gboolean use_valgrind = FALSE;
gboolean use_callgrind = FALSE;
const char *devnull = "/dev/null";
const char *env_valgrind = getenv("PCMK_valgrind_enabled");
const char *env_callgrind = getenv("PCMK_callgrind_enabled");
enum cluster_type_e stack = get_cluster_type();
child->active_before_startup = FALSE;
if (child->command == NULL) {
crm_info("Nothing to do for child \"%s\"", child->name);
return TRUE;
}
if (env_callgrind != NULL && crm_is_true(env_callgrind)) {
use_callgrind = TRUE;
use_valgrind = TRUE;
} else if (env_callgrind != NULL && strstr(env_callgrind, child->name)) {
use_callgrind = TRUE;
use_valgrind = TRUE;
} else if (env_valgrind != NULL && crm_is_true(env_valgrind)) {
use_valgrind = TRUE;
} else if (env_valgrind != NULL && strstr(env_valgrind, child->name)) {
use_valgrind = TRUE;
}
if (use_valgrind && strlen(VALGRIND_BIN) == 0) {
crm_warn("Cannot enable valgrind for %s:"
" The location of the valgrind binary is unknown", child->name);
use_valgrind = FALSE;
}
if (child->uid) {
if (crm_user_lookup(child->uid, &uid, &gid) < 0) {
crm_err("Invalid user (%s) for %s: not found", child->uid, child->name);
return FALSE;
}
crm_info("Using uid=%u and group=%u for process %s", uid, gid, child->name);
}
child->pid = fork();
CRM_ASSERT(child->pid != -1);
if (child->pid > 0) {
/* parent */
mainloop_child_add(child->pid, 0, child->name, child, pcmk_child_exit);
crm_info("Forked child %d for process %s%s", child->pid, child->name,
use_valgrind ? " (valgrind enabled: " VALGRIND_BIN ")" : "");
update_node_processes(local_nodeid, NULL, get_process_list());
return TRUE;
} else {
/* Start a new session */
(void)setsid();
/* Setup the two alternate arg arrarys */
opts_vgrind[0] = strdup(VALGRIND_BIN);
if (use_callgrind) {
opts_vgrind[1] = strdup("--tool=callgrind");
opts_vgrind[2] = strdup("--callgrind-out-file=" CRM_STATE_DIR "/callgrind.out.%p");
opts_vgrind[3] = strdup(child->command);
opts_vgrind[4] = NULL;
} else {
opts_vgrind[1] = strdup(child->command);
opts_vgrind[2] = NULL;
opts_vgrind[3] = NULL;
opts_vgrind[4] = NULL;
}
opts_default[0] = strdup(child->command);;
if(gid) {
if(stack == pcmk_cluster_corosync) {
/* Drop root privileges completely
*
* We can do this because we set uidgid.gid.${gid}=1
* via CMAP which allows these processes to connect to
* corosync
*/
if (setgid(gid) < 0) {
crm_perror(LOG_ERR, "Could not set group to %d", gid);
}
/* Keep the root group (so we can access corosync), but add the haclient group (so we can access ipc) */
} else if (initgroups(child->uid, gid) < 0) {
crm_err("Cannot initialize groups for %s: %s (%d)", child->uid, pcmk_strerror(errno), errno);
}
}
if (uid && setuid(uid) < 0) {
crm_perror(LOG_ERR, "Could not set user to %d (%s)", uid, child->uid);
}
/* Close all open file descriptors */
getrlimit(RLIMIT_NOFILE, &oflimits);
for (lpc = 0; 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 (use_valgrind) {
(void)execvp(VALGRIND_BIN, opts_vgrind);
} else {
(void)execvp(child->command, opts_default);
}
crm_perror(LOG_ERR, "FATAL: Cannot exec %s", child->command);
crm_exit(DAEMON_RESPAWN_STOP);
}
return TRUE; /* never reached */
}
static gboolean
escalate_shutdown(gpointer data)
{
pcmk_child_t *child = data;
if (child->pid) {
/* Use SIGSEGV instead of SIGKILL to create a core so we can see what it was up to */
crm_err("Child %s not terminating in a timely manner, forcing", child->name);
stop_child(child, SIGSEGV);
}
return FALSE;
}
static gboolean
pcmk_shutdown_worker(gpointer user_data)
{
static int phase = 0;
static time_t next_log = 0;
static int max = SIZEOF(pcmk_children);
int lpc = 0;
if (phase == 0) {
crm_notice("Shuting down Pacemaker");
phase = max;
/* Add a second, more frequent, check to speed up shutdown */
g_timeout_add_seconds(5, check_active_before_startup_processes, NULL);
}
for (; phase > 0; phase--) {
/* dont stop anything with start_seq < 1 */
for (lpc = max - 1; lpc >= 0; lpc--) {
pcmk_child_t *child = &(pcmk_children[lpc]);
if (phase != child->start_seq) {
continue;
}
if (child->pid) {
time_t now = time(NULL);
if (child->respawn) {
next_log = now + 30;
child->respawn = FALSE;
stop_child(child, SIGTERM);
if (phase < pcmk_children[pcmk_child_crmd].start_seq) {
g_timeout_add(180000 /* 3m */ , escalate_shutdown, child);
}
} else if (now >= next_log) {
next_log = now + 30;
crm_notice("Still waiting for %s (pid=%d, seq=%d) to terminate...",
child->name, child->pid, child->start_seq);
}
return TRUE;
}
/* cleanup */
crm_debug("%s confirmed stopped", child->name);
child->pid = 0;
}
}
/* send_cluster_id(); */
crm_notice("Shutdown complete");
{
const char *delay = daemon_option("shutdown_delay");
if(delay) {
sync();
sleep(crm_get_msec(delay) / 1000);
}
}
g_main_loop_quit(mainloop);
if (fatal_error) {
crm_notice("Attempting to inhibit respawning after fatal error");
crm_exit(DAEMON_RESPAWN_STOP);
}
return TRUE;
}
static void
pcmk_ignore(int nsig)
{
crm_info("Ignoring signal %s (%d)", strsignal(nsig), nsig);
}
static void
pcmk_sigquit(int nsig)
{
pcmk_panic(__FUNCTION__);
}
void
pcmk_shutdown(int nsig)
{
if (shutdown_trigger == NULL) {
shutdown_trigger = mainloop_add_trigger(G_PRIORITY_HIGH, pcmk_shutdown_worker, NULL);
}
mainloop_set_trigger(shutdown_trigger);
}
static int32_t
pcmk_ipc_accept(qb_ipcs_connection_t * c, uid_t uid, gid_t gid)
{
crm_trace("Connection %p", c);
if (crm_client_new(c, uid, gid) == NULL) {
return -EIO;
}
return 0;
}
static void
pcmk_ipc_created(qb_ipcs_connection_t * c)
{
crm_trace("Connection %p", c);
}
/* Exit code means? */
static int32_t
pcmk_ipc_dispatch(qb_ipcs_connection_t * qbc, void *data, size_t size)
{
uint32_t id = 0;
uint32_t flags = 0;
const char *task = NULL;
crm_client_t *c = crm_client_get(qbc);
xmlNode *msg = crm_ipcs_recv(c, data, size, &id, &flags);
crm_ipcs_send_ack(c, id, flags, "ack", __FUNCTION__, __LINE__);
if (msg == NULL) {
return 0;
}
task = crm_element_value(msg, F_CRM_TASK);
if (crm_str_eq(task, CRM_OP_QUIT, TRUE)) {
/* Time to quit */
crm_notice("Shutting down in responce to ticket %s (%s)",
crm_element_value(msg, F_CRM_REFERENCE), crm_element_value(msg, F_CRM_ORIGIN));
pcmk_shutdown(15);
} else if (crm_str_eq(task, CRM_OP_RM_NODE_CACHE, TRUE)) {
/* Send to everyone */
struct iovec *iov;
int id = 0;
const char *name = NULL;
crm_element_value_int(msg, XML_ATTR_ID, &id);
name = crm_element_value(msg, XML_ATTR_UNAME);
crm_notice("Instructing peers to remove references to node %s/%u", name, id);
iov = calloc(1, sizeof(struct iovec));
iov->iov_base = dump_xml_unformatted(msg);
iov->iov_len = 1 + strlen(iov->iov_base);
send_cpg_iov(iov);
} else {
update_process_clients(c);
}
free_xml(msg);
return 0;
}
/* Error code means? */
static int32_t
pcmk_ipc_closed(qb_ipcs_connection_t * c)
{
crm_client_t *client = crm_client_get(c);
if (client == NULL) {
return 0;
}
crm_trace("Connection %p", c);
crm_client_destroy(client);
return 0;
}
static void
pcmk_ipc_destroy(qb_ipcs_connection_t * c)
{
crm_trace("Connection %p", c);
pcmk_ipc_closed(c);
}
struct qb_ipcs_service_handlers mcp_ipc_callbacks = {
.connection_accept = pcmk_ipc_accept,
.connection_created = pcmk_ipc_created,
.msg_process = pcmk_ipc_dispatch,
.connection_closed = pcmk_ipc_closed,
.connection_destroyed = pcmk_ipc_destroy
};
/*!
* \internal
* \brief Send an XML message with process list of all known peers to client(s)
*
* \param[in] client Send message to this client, or all clients if NULL
*/
void
update_process_clients(crm_client_t *client)
{
GHashTableIter iter;
crm_node_t *node = NULL;
xmlNode *update = create_xml_node(NULL, "nodes");
g_hash_table_iter_init(&iter, crm_peer_cache);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) & node)) {
xmlNode *xml = create_xml_node(update, "node");
crm_xml_add_int(xml, "id", node->id);
crm_xml_add(xml, "uname", node->uname);
crm_xml_add(xml, "state", node->state);
crm_xml_add_int(xml, "processes", node->processes);
}
if(client) {
crm_trace("Sending process list to client %s", client->id);
crm_ipcs_send(client, 0, update, crm_ipc_server_event);
} else {
crm_trace("Sending process list to %d clients", crm_hash_table_size(client_connections));
g_hash_table_iter_init(&iter, client_connections);
while (g_hash_table_iter_next(&iter, NULL, (gpointer *) & client)) {
crm_ipcs_send(client, 0, update, crm_ipc_server_event);
}
}
free_xml(update);
}
/*!
* \internal
* \brief Send a CPG message with local node's process list to all peers
*/
void
update_process_peers(void)
{
/* Do nothing for corosync-2 based clusters */
char buffer[1024];
struct iovec *iov;
int rc = 0;
memset(buffer, 0, SIZEOF(buffer));
if (local_name) {
rc = snprintf(buffer, SIZEOF(buffer) - 1, "<node uname=\"%s\" proclist=\"%u\"/>",
local_name, get_process_list());
} else {
rc = snprintf(buffer, SIZEOF(buffer) - 1, "<node proclist=\"%u\"/>", get_process_list());
}
crm_trace("Sending %s", buffer);
iov = calloc(1, sizeof(struct iovec));
iov->iov_base = strdup(buffer);
iov->iov_len = rc + 1;
send_cpg_iov(iov);
}
/*!
* \internal
* \brief Update a node's process list, notifying clients and peers if needed
*
* \param[in] id Node ID of affected node
* \param[in] uname Uname of affected node
* \param[in] procs Affected node's process list mask
*
* \return TRUE if the process list changed, FALSE otherwise
*/
gboolean
update_node_processes(uint32_t id, const char *uname, uint32_t procs)
{
gboolean changed = FALSE;
crm_node_t *node = crm_get_peer(id, uname);
if (procs != 0) {
if (procs != node->processes) {
crm_debug("Node %s now has process list: %.32x (was %.32x)",
node->uname, procs, node->processes);
node->processes = procs;
changed = TRUE;
/* If local node's processes have changed, notify clients/peers */
if (id == local_nodeid) {
update_process_clients(NULL);
update_process_peers();
}
} else {
crm_trace("Node %s still has process list: %.32x", node->uname, procs);
}
}
return changed;
}
/* *INDENT-OFF* */
static struct crm_option long_options[] = {
/* Top-level Options */
{"help", 0, 0, '?', "\tThis text"},
{"version", 0, 0, '$', "\tVersion information" },
{"verbose", 0, 0, 'V', "\tIncrease debug output"},
{"shutdown", 0, 0, 'S', "\tInstruct Pacemaker to shutdown on this machine"},
{"features", 0, 0, 'F', "\tDisplay the full version and list of features Pacemaker was built with"},
{"-spacer-", 1, 0, '-', "\nAdditional Options:"},
{"foreground", 0, 0, 'f', "\t(Ignored) Pacemaker always runs in the foreground"},
{"pid-file", 1, 0, 'p', "\t(Ignored) Daemon pid file location"},
{NULL, 0, 0, 0}
};
/* *INDENT-ON* */
static void
mcp_chown(const char *path, uid_t uid, gid_t gid)
{
int rc = chown(path, uid, gid);
if (rc < 0) {
crm_warn("Cannot change the ownership of %s to user %s and gid %d: %s",
path, CRM_DAEMON_USER, gid, pcmk_strerror(errno));
}
}
static gboolean
check_active_before_startup_processes(gpointer user_data)
{
int start_seq = 1, lpc = 0;
static int max = SIZEOF(pcmk_children);
gboolean keep_tracking = FALSE;
for (start_seq = 1; start_seq < max; start_seq++) {
for (lpc = 0; lpc < max; lpc++) {
if (pcmk_children[lpc].active_before_startup == FALSE) {
/* we are already tracking it as a child process. */
continue;
} else if (start_seq != pcmk_children[lpc].start_seq) {
continue;
} else if (crm_pid_active(pcmk_children[lpc].pid) != 1) {
crm_notice("Process %s terminated (pid=%d)",
pcmk_children[lpc].name, pcmk_children[lpc].pid);
pcmk_process_exit(&(pcmk_children[lpc]));
continue;
}
/* at least one of the processes found at startup
* is still going, so keep this recurring timer around */
keep_tracking = TRUE;
}
}
return keep_tracking;
}
static bool
find_and_track_existing_processes(void)
{
DIR *dp;
struct dirent *entry;
struct stat statbuf;
int start_tracker = 0;
dp = opendir("/proc");
if (!dp) {
/* no proc directory to search through */
crm_notice("Can not read /proc directory to track existing components");
return FALSE;
}
while ((entry = readdir(dp)) != NULL) {
char procpath[128];
char value[64];
char key[16];
FILE *file;
int pid;
int max = SIZEOF(pcmk_children);
int i;
strcpy(procpath, "/proc/");
/* strlen("/proc/") + strlen("/status") + 1 = 14
* 128 - 14 = 114 */
strncat(procpath, entry->d_name, 114);
if (lstat(procpath, &statbuf)) {
continue;
}
if (!S_ISDIR(statbuf.st_mode) || !isdigit(entry->d_name[0])) {
continue;
}
strcat(procpath, "/status");
file = fopen(procpath, "r");
if (!file) {
continue;
}
if (fscanf(file, "%15s%63s", key, value) != 2) {
fclose(file);
continue;
}
fclose(file);
pid = atoi(entry->d_name);
if (pid <= 0) {
continue;
}
for (i = 0; i < max; i++) {
const char *name = pcmk_children[i].name;
if (pcmk_children[i].start_seq == 0) {
continue;
}
if (pcmk_children[i].flag == crm_proc_stonith_ng) {
name = "stonithd";
}
if (safe_str_eq(name, value)) {
if (crm_pid_active(pid) != 1) {
continue;
}
crm_notice("Tracking existing %s process (pid=%d)", value, pid);
pcmk_children[i].pid = pid;
pcmk_children[i].active_before_startup = TRUE;
start_tracker = 1;
}
}
}
if (start_tracker) {
g_timeout_add_seconds(PCMK_PROCESS_CHECK_INTERVAL, check_active_before_startup_processes,
NULL);
}
closedir(dp);
return start_tracker;
}
static void
init_children_processes(void)
{
int start_seq = 1, lpc = 0;
static int max = SIZEOF(pcmk_children);
/* start any children that have not been detected */
for (start_seq = 1; start_seq < max; start_seq++) {
/* dont start anything with start_seq < 1 */
for (lpc = 0; lpc < max; lpc++) {
if (pcmk_children[lpc].pid) {
/* we are already tracking it */
continue;
}
if (start_seq == pcmk_children[lpc].start_seq) {
start_child(&(pcmk_children[lpc]));
}
}
}
/* From this point on, any daemons being started will be due to
* respawning rather than node start.
*
* This may be useful for the daemons to know
*/
setenv("PCMK_respawned", "true", 1);
}
static void
mcp_cpg_destroy(gpointer user_data)
{
crm_err("Connection destroyed");
crm_exit(ENOTCONN);
}
/*!
* \internal
* \brief Process a CPG message (process list or manual peer cache removal)
*
* \param[in] handle CPG connection (ignored)
* \param[in] groupName CPG group name (ignored)
* \param[in] nodeid ID of affected node
* \param[in] pid Process ID (ignored)
* \param[in] msg CPG XML message
* \param[in] msg_len Length of msg in bytes (ignored)
*/
static void
mcp_cpg_deliver(cpg_handle_t handle,
const struct cpg_name *groupName,
uint32_t nodeid, uint32_t pid, void *msg, size_t msg_len)
{
xmlNode *xml = string2xml(msg);
const char *task = crm_element_value(xml, F_CRM_TASK);
crm_trace("Received CPG message (%s): %.200s",
(task? task : "process list"), msg);
if (task == NULL) {
if (nodeid == local_nodeid) {
crm_info("Ignoring process list sent by peer for local node");
} else {
uint32_t procs = 0;
const char *uname = crm_element_value(xml, "uname");
crm_element_value_int(xml, "proclist", (int *)&procs);
if (update_node_processes(nodeid, uname, procs)) {
update_process_clients(NULL);
}
}
} else if (crm_str_eq(task, CRM_OP_RM_NODE_CACHE, TRUE)) {
int id = 0;
const char *name = NULL;
crm_element_value_int(xml, XML_ATTR_ID, &id);
name = crm_element_value(xml, XML_ATTR_UNAME);
reap_crm_member(id, name);
}
if (xml != NULL) {
free_xml(xml);
}
}
static void
mcp_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)
{
- /* Don't care about CPG membership, but we do want to broadcast our own presence */
+ /* Update peer cache if needed */
+ pcmk_cpg_membership(handle, groupName, member_list, member_list_entries,
+ left_list, left_list_entries,
+ joined_list, joined_list_entries);
+
+ /* Always broadcast our own presence after any membership change */
update_process_peers();
}
static gboolean
mcp_quorum_callback(unsigned long long seq, gboolean quorate)
{
/* Nothing to do */
return TRUE;
}
static void
mcp_quorum_destroy(gpointer user_data)
{
crm_info("connection closed");
}
int
main(int argc, char **argv)
{
int rc;
int flag;
int argerr = 0;
int option_index = 0;
gboolean shutdown = FALSE;
uid_t pcmk_uid = 0;
gid_t pcmk_gid = 0;
struct rlimit cores;
crm_ipc_t *old_instance = NULL;
qb_ipcs_service_t *ipcs = NULL;
const char *facility = daemon_option("logfacility");
static crm_cluster_t cluster;
crm_log_preinit(NULL, argc, argv);
crm_set_options(NULL, "mode [options]", long_options, "Start/Stop Pacemaker\n");
mainloop_add_signal(SIGHUP, pcmk_ignore);
mainloop_add_signal(SIGQUIT, pcmk_sigquit);
while (1) {
flag = crm_get_option(argc, argv, &option_index);
if (flag == -1)
break;
switch (flag) {
case 'V':
crm_bump_log_level(argc, argv);
break;
case 'f':
/* Legacy */
break;
case 'p':
pid_file = optarg;
break;
case '$':
case '?':
crm_help(flag, EX_OK);
break;
case 'S':
shutdown = TRUE;
break;
case 'F':
printf("Pacemaker %s (Build: %s)\n Supporting v%s: %s\n", VERSION, BUILD_VERSION,
CRM_FEATURE_SET, CRM_FEATURES);
crm_exit(pcmk_ok);
default:
printf("Argument code 0%o (%c) is not (?yet?) supported\n", flag, flag);
++argerr;
break;
}
}
if (optind < argc) {
printf("non-option ARGV-elements: ");
while (optind < argc)
printf("%s ", argv[optind++]);
printf("\n");
}
if (argerr) {
crm_help('?', EX_USAGE);
}
setenv("LC_ALL", "C", 1);
setenv("HA_LOGD", "no", 1);
set_daemon_option("mcp", "true");
set_daemon_option("use_logd", "off");
crm_log_init(NULL, LOG_INFO, TRUE, FALSE, argc, argv, FALSE);
/* Restore the original facility so that mcp_read_config() does the right thing */
set_daemon_option("logfacility", facility);
crm_debug("Checking for old instances of %s", CRM_SYSTEM_MCP);
old_instance = crm_ipc_new(CRM_SYSTEM_MCP, 0);
crm_ipc_connect(old_instance);
if (shutdown) {
crm_debug("Terminating previous instance");
while (crm_ipc_connected(old_instance)) {
xmlNode *cmd =
create_request(CRM_OP_QUIT, NULL, NULL, CRM_SYSTEM_MCP, CRM_SYSTEM_MCP, NULL);
crm_debug(".");
crm_ipc_send(old_instance, cmd, 0, 0, NULL);
free_xml(cmd);
sleep(2);
}
crm_ipc_close(old_instance);
crm_ipc_destroy(old_instance);
crm_exit(pcmk_ok);
} else if (crm_ipc_connected(old_instance)) {
crm_ipc_close(old_instance);
crm_ipc_destroy(old_instance);
crm_err("Pacemaker is already active, aborting startup");
crm_exit(DAEMON_RESPAWN_STOP);
}
crm_ipc_close(old_instance);
crm_ipc_destroy(old_instance);
if (mcp_read_config() == FALSE) {
crm_notice("Could not obtain corosync config data, exiting");
crm_exit(ENODATA);
}
crm_notice("Starting Pacemaker %s (Build: %s): %s", VERSION, BUILD_VERSION, CRM_FEATURES);
mainloop = g_main_new(FALSE);
sysrq_init();
rc = getrlimit(RLIMIT_CORE, &cores);
if (rc < 0) {
crm_perror(LOG_ERR, "Cannot determine current maximum core size.");
} else {
if (cores.rlim_max == 0 && geteuid() == 0) {
cores.rlim_max = RLIM_INFINITY;
} else {
crm_info("Maximum core file size is: %lu", (unsigned long)cores.rlim_max);
}
cores.rlim_cur = cores.rlim_max;
rc = setrlimit(RLIMIT_CORE, &cores);
if (rc < 0) {
crm_perror(LOG_ERR,
"Core file generation will remain disabled."
" Core files are an important diagnositic tool,"
" please consider enabling them by default.");
}
#if 0
/* system() is not thread-safe, can't call from here
* Actually, its a pretty hacky way to try and achieve this anyway
*/
if (system("echo 1 > /proc/sys/kernel/core_uses_pid") != 0) {
crm_perror(LOG_ERR, "Could not enable /proc/sys/kernel/core_uses_pid");
}
#endif
}
rc = pcmk_ok;
if (crm_user_lookup(CRM_DAEMON_USER, &pcmk_uid, &pcmk_gid) < 0) {
crm_err("Cluster user %s does not exist, aborting Pacemaker startup", CRM_DAEMON_USER);
crm_exit(ENOKEY);
}
mkdir(CRM_STATE_DIR, 0750);
mcp_chown(CRM_STATE_DIR, pcmk_uid, pcmk_gid);
/* Used to store core files in */
crm_build_path(CRM_CORE_DIR, 0775);
mcp_chown(CRM_CORE_DIR, pcmk_uid, pcmk_gid);
/* Used to store blackbox dumps in */
crm_build_path(CRM_BLACKBOX_DIR, 0755);
mcp_chown(CRM_BLACKBOX_DIR, pcmk_uid, pcmk_gid);
/* Used to store policy engine inputs in */
crm_build_path(PE_STATE_DIR, 0755);
mcp_chown(PE_STATE_DIR, pcmk_uid, pcmk_gid);
/* Used to store the cluster configuration */
crm_build_path(CRM_CONFIG_DIR, 0755);
mcp_chown(CRM_CONFIG_DIR, pcmk_uid, pcmk_gid);
/* Resource agent paths are constructed by the lrmd */
ipcs = mainloop_add_ipc_server(CRM_SYSTEM_MCP, QB_IPC_NATIVE, &mcp_ipc_callbacks);
if (ipcs == NULL) {
crm_err("Couldn't start IPC server");
crm_exit(EIO);
}
/* Allows us to block shutdown */
if (cluster_connect_cfg(&local_nodeid) == FALSE) {
crm_err("Couldn't connect to Corosync's CFG service");
crm_exit(ENOPROTOOPT);
}
if(pcmk_locate_sbd() > 0) {
setenv("PCMK_watchdog", "true", 1);
} else {
setenv("PCMK_watchdog", "false", 1);
}
find_and_track_existing_processes();
cluster.destroy = mcp_cpg_destroy;
cluster.cpg.cpg_deliver_fn = mcp_cpg_deliver;
cluster.cpg.cpg_confchg_fn = mcp_cpg_membership;
if(cluster_connect_cpg(&cluster) == FALSE) {
crm_err("Couldn't connect to Corosync's CPG service");
rc = -ENOPROTOOPT;
}
if (rc == pcmk_ok && is_corosync_cluster()) {
/* Keep the membership list up-to-date for crm_node to query */
if(cluster_connect_quorum(mcp_quorum_callback, mcp_quorum_destroy) == FALSE) {
rc = -ENOTCONN;
}
}
if(rc == pcmk_ok) {
local_name = get_local_node_name();
update_node_processes(local_nodeid, local_name, get_process_list());
mainloop_add_signal(SIGTERM, pcmk_shutdown);
mainloop_add_signal(SIGINT, pcmk_shutdown);
init_children_processes();
crm_info("Starting mainloop");
g_main_run(mainloop);
}
if (ipcs) {
crm_trace("Closing IPC server");
mainloop_del_ipc_server(ipcs);
ipcs = NULL;
}
g_main_destroy(mainloop);
cluster_disconnect_cpg(&cluster);
cluster_disconnect_cfg();
crm_info("Exiting %s", crm_system_name);
return crm_exit(rc);
}

File Metadata

Mime Type
text/x-diff
Expires
Tue, Jul 8, 6:28 PM (10 h, 55 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
2001223
Default Alt Text
(263 KB)

Event Timeline