diff --git a/attrd/Makefile.am b/attrd/Makefile.am index 6eaaae2628..f72cf19c65 100644 --- a/attrd/Makefile.am +++ b/attrd/Makefile.am @@ -1,40 +1,42 @@ # # Copyright (C) 2004-2009 Andrew Beekhof # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # include $(top_srcdir)/Makefile.common halibdir = $(CRM_DAEMON_DIR) halib_PROGRAMS = attrd ## SOURCES +noinst_HEADERS = internal.h attrd_common.h + attrd_CFLAGS = $(CFLAGS_HARDENED_EXE) attrd_LDFLAGS = $(LDFLAGS_HARDENED_EXE) attrd_LDADD = $(top_builddir)/lib/cluster/libcrmcluster.la \ $(top_builddir)/lib/common/libcrmcommon.la \ $(top_builddir)/lib/cib/libcib.la \ $(CLUSTERLIBS) attrd_SOURCES = if BUILD_ATOMIC_ATTRD -attrd_SOURCES += main.c commands.c +attrd_SOURCES += main.c commands.c attrd_common.c else -attrd_SOURCES += legacy.c +attrd_SOURCES += legacy.c attrd_common.c endif clean-generic: rm -f *.log *.debug *.xml *~ diff --git a/attrd/attrd_common.c b/attrd/attrd_common.c new file mode 100644 index 0000000000..49bea9daaa --- /dev/null +++ b/attrd/attrd_common.c @@ -0,0 +1,242 @@ +/* + * Copyright (C) 2004-2017 Andrew Beekhof + * + * This source code is licensed under the GNU General Public License version 2 + * or later (GPLv2+) WITHOUT ANY WARRANTY. + */ + +#include + +#include +#include +#include +#include + +#include +#include +#include + +#include + +static gboolean shutting_down = FALSE; +static GMainLoop *mloop = NULL; + +/*! + * \internal + * \brief Check whether we're currently shutting down + * + * \return TRUE if shutting down, FALSE otherwise + */ +gboolean +attrd_shutting_down() +{ + return shutting_down; +} + +/*! + * \internal + * \brief Exit (using mainloop or not, as appropriate) + * + * \param[in] nsig Ignored + */ +void +attrd_shutdown(int nsig) { + crm_info("Shutting down"); + + shutting_down = TRUE; + + if ((mloop != NULL) && g_main_is_running(mloop)) { + g_main_quit(mloop); + } else { + crm_exit(pcmk_ok); + } +} + +/*! + * \internal + * \brief Create a main loop for attrd + */ +void +attrd_init_mainloop() +{ + mloop = g_main_new(FALSE); +} + +/*! + * \internal + * \brief Run attrd main loop + */ +void +attrd_run_mainloop() +{ + g_main_run(mloop); +} + +/*! + * \internal + * \brief Check whether attrd main loop is running + * + * \return TRUE if main loop is running, FALSE otherwise + */ +gboolean +attrd_mainloop_running() +{ + return (mloop != NULL) && g_main_is_running(mloop); +} + +/*! + * \internal + * \brief Quit attrd mainloop + */ +void +attrd_quit_mainloop() +{ + g_main_quit(mloop); +} + +/*! + * \internal + * \brief Accept a new client IPC connection + * + * \param[in] c New connection + * \param[in] uid Client user id + * \param[in] gid Client group id + * + * \return pcmk_ok on success, -errno otherwise + */ +static int32_t +attrd_ipc_accept(qb_ipcs_connection_t *c, uid_t uid, gid_t gid) +{ + crm_trace("New client connection %p", c); + if (shutting_down) { + crm_info("Ignoring new connection from pid %d during shutdown", + crm_ipcs_client_pid(c)); + return -EPERM; + } + + if (crm_client_new(c, uid, gid) == NULL) { + return -EIO; + } + return pcmk_ok; +} + +/*! + * \internal + * \brief Callback for successful client connection + * + * \param[in] c New connection + */ +static void +attrd_ipc_created(qb_ipcs_connection_t *c) +{ + crm_trace("Client connection %p accepted", c); +} + +/*! + * \internal + * \brief Destroy a client IPC connection + * + * \param[in] c Connection to destroy + * + * \return FALSE (i.e. do not re-run this callback) + */ +static int32_t +attrd_ipc_closed(qb_ipcs_connection_t *c) +{ + crm_client_t *client = crm_client_get(c); + + if (client == NULL) { + crm_trace("Ignoring request to clean up unknown connection %p", c); + } else { + crm_trace("Cleaning up closed client connection %p", c); + crm_client_destroy(client); + } + return FALSE; +} + +/*! + * \internal + * \brief Destroy a client IPC connection + * + * \param[in] c Connection to destroy + * + * \note We handle a destroyed connection the same as a closed one, + * but we need a separate handler because the return type is different. + */ +static void +attrd_ipc_destroy(qb_ipcs_connection_t *c) +{ + crm_trace("Destroying client connection %p", c); + attrd_ipc_closed(c); +} + +/*! + * \internal + * \brief Set up attrd IPC communication + * + * \param[out] ipcs Will be set to newly allocated server connection + * \param[in] dispatch_fn Handler for new messages on connection + */ +void +attrd_init_ipc(qb_ipcs_service_t **ipcs, qb_ipcs_msg_process_fn dispatch_fn) +{ + + static struct qb_ipcs_service_handlers ipc_callbacks = { + .connection_accept = attrd_ipc_accept, + .connection_created = attrd_ipc_created, + .msg_process = NULL, + .connection_closed = attrd_ipc_closed, + .connection_destroyed = attrd_ipc_destroy + }; + + ipc_callbacks.msg_process = dispatch_fn; + attrd_ipc_server_init(ipcs, &ipc_callbacks); +} + +/* strlen("value") */ +#define plus_plus_len (5) + +/*! + * \internal + * \brief Check whether an attribute value should be expanded + * + * \param[in] value Attribute value to check + * + * \return TRUE if value needs expansion, FALSE otherwise + */ +gboolean +attrd_value_needs_expansion(const char *value) +{ + return ((strlen(value) >= (plus_plus_len + 2)) + && (value[plus_plus_len] == '+') + && ((value[plus_plus_len + 1] == '+') + || (value[plus_plus_len + 1] == '='))); +} + +/*! + * \internal + * \brief Expand an increment expression into an integer + * + * \param[in] value Attribute increment expression to expand + * \param[in] old_value Previous value of attribute + * + * \return Expanded value + */ +int +attrd_expand_value(const char *value, const char *old_value) +{ + int offset = 1; + int int_value = char2score(old_value); + + 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; + } + return int_value; +} diff --git a/attrd/attrd_common.h b/attrd/attrd_common.h new file mode 100644 index 0000000000..c3f0f194a7 --- /dev/null +++ b/attrd/attrd_common.h @@ -0,0 +1,24 @@ +/* + * Copyright (C) 2017 Andrew Beekhof + * + * This source code is licensed under the GNU General Public License version 2 + * or later (GPLv2+) WITHOUT ANY WARRANTY. + */ + +#ifndef PCMK_ATTRD_COMMON__H +# define PCMK_ATTRD_COMMON__H + +void attrd_init_mainloop(void); +void attrd_run_mainloop(void); +gboolean attrd_mainloop_running(void); +void attrd_quit_mainloop(void); + +gboolean attrd_shutting_down(void); +void attrd_shutdown(int nsig); +void attrd_init_ipc(qb_ipcs_service_t **ipcs, + qb_ipcs_msg_process_fn dispatch_fn); + +gboolean attrd_value_needs_expansion(const char *value); +int attrd_expand_value(const char *value, const char *old_value); + +#endif /* PCMK_ATTRD_COMMON__H */ diff --git a/attrd/commands.c b/attrd/commands.c index 83502d430c..82a975cdc5 100644 --- a/attrd/commands.c +++ b/attrd/commands.c @@ -1,1094 +1,1077 @@ /* * Copyright (C) 2013 Andrew Beekhof * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU General Public * License as published by the Free Software Foundation; either * version 2 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 #include #include #include #include #include #include #include #include #include #include #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; } 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(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); } 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); } else if (dampen < 0) { crm_warn("Ignoring invalid delay %s for attribute %s", value, a->id); } 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; } else if (attr == NULL) { crm_err("Update request did not specify attribute or regular expression"); free(key); free(set); free(host); 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 (attrd_value_needs_expansion(value)) { + int int_value; + attribute_value_t *v = NULL; 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; - } + int_value = attrd_expand_value(value, (v? v->current : NULL)); crm_info("Expanded %s=%s to %d", attr, value, int_value); crm_xml_add_int(xml, F_ATTRD_VALUE, int_value); /* Replacing the value frees the previous memory, so re-query it */ value = crm_element_value(xml, F_ATTRD_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) || safe_str_eq(op, ATTRD_OP_UPDATE_BOTH) || safe_str_eq(op, ATTRD_OP_UPDATE_DELAY)) { 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)) { attrd_peer_remove(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_info("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); } /*! * \internal * \brief Remove all attributes and optionally peer cache entries for a node * * \param[in] host Name of node to purge * \param[in] uncache If TRUE, remove node from peer caches * \param[in] source Who requested removal (only used for logging) */ void attrd_peer_remove(const char *host, gboolean uncache, const char *source) { attribute_t *a = NULL; GHashTableIter aIter; CRM_CHECK(host != NULL, return); crm_notice("Removing all %s attributes for %s", host, source); 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(0, 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); int is_remote = 0; crm_element_value_int(xml, F_ATTRD_IS_REMOTE, &is_remote); if (is_remote) { /* If we previously assumed this node was an unseen cluster node, * remove its entry from the cluster peer cache. */ crm_node_t *dup = crm_find_peer(0, host); if (dup && (dup->uuid == NULL)) { reap_crm_member(0, host); } /* Ensure this host is in the remote peer cache */ crm_remote_peer_cache_add(host); } if (v == NULL) { v = calloc(1, sizeof(attribute_value_t)); CRM_ASSERT(v != NULL); v->nodename = strdup(host); CRM_ASSERT(v->nodename != NULL); v->is_remote = is_remote; 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_t *a; attribute_value_t *v = NULL; int dampen = 0; const char *op = crm_element_value(xml, F_ATTRD_TASK); const char *attr = crm_element_value(xml, F_ATTRD_ATTRIBUTE); const char *value = crm_element_value(xml, F_ATTRD_VALUE); const char *dvalue = crm_element_value(xml, F_ATTRD_DAMPEN); if (attr == NULL) { crm_warn("Peer update did not specify attribute"); return; } a = g_hash_table_lookup(attributes, attr); if(a == NULL) { if (op == NULL /* The xml children from an ATTRD_OP_SYNC_RESPONSE have no F_ATTRD_TASK */ || safe_str_eq(op, ATTRD_OP_UPDATE) || safe_str_eq(op, ATTRD_OP_UPDATE_BOTH)) { a = create_attribute(xml); } else { crm_warn("Update error (attribute %s not found)", attr); return; } } if (op == NULL /* The xml children from an ATTRD_OP_SYNC_RESPONSE have no F_ATTRD_TASK */ || safe_str_eq(op, ATTRD_OP_UPDATE_BOTH) || safe_str_eq(op, ATTRD_OP_UPDATE_DELAY)) { if (dvalue) { dampen = crm_get_msec(dvalue); if (dampen >= 0) { if (a->timeout_ms != dampen) { mainloop_timer_stop(a->timer); mainloop_timer_del(a->timer); a->timeout_ms = dampen; if (dampen > 0) { a->timer = mainloop_timer_add(a->id, a->timeout_ms, FALSE, attribute_timer_cb, a); crm_info("Update attribute %s with delay %dms (%s)", a->id, dampen, dvalue); } else { a->timer = NULL; crm_info("Update attribute %s with not delay", a->id); } //if dampen is changed, attrd writes in a current value immediately. write_or_elect_attribute(a); if (safe_str_eq(op, ATTRD_OP_UPDATE_DELAY)) { return; } } else { if (safe_str_eq(op, ATTRD_OP_UPDATE_DELAY)) { crm_trace("Unchanged attribute %s with delay %dms (%s).(ATTRD_OP_UPDATE_DELAY)", a->id, dampen, dvalue); return; } } } else { crm_warn("Update error (A positive number is necessary for delay parameter. attribute %s : %dms (%s))", a->id, dampen, dvalue); return; } } else { crm_warn("Update error (delay parameter is necessary for the update of the attribute %s)", a->id); return; } } 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); } } /* If this is a cluster node whose node ID we are learning, remember it */ if ((v->nodeid == 0) && (v->is_remote == FALSE) && (crm_element_value_int(xml, F_ATTRD_HOST_ID, (int*)&v->nodeid) == 0)) { crm_node_t *known_peer = crm_get_peer(v->nodeid, host); crm_trace("We know %s's node id now: %s", known_peer->uname, known_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 we're the writer, send new peers a list of all attributes * (unless it's a remote node, which doesn't run its own attrd) */ if ((election_state(writer) == election_won) && !is_set(peer->flags, crm_remote_node)) { attrd_peer_sync(peer, NULL); } } else { /* Remove all attribute values associated with lost nodes */ attrd_peer_remove(peer->uname, FALSE, "peer loss"); if (peer_writer && safe_str_eq(peer->uname, peer_writer)) { free(peer_writer); peer_writer = NULL; crm_notice("Lost attribute writer %s", peer->uname); } } } } 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); free(v->requested); v->requested = NULL; if (rc != pcmk_ok) { a->changed = TRUE; /* Attempt write out again */ } } done: 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_ANY); /* 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 : ""), 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 : ""), a->set); the_cib->cmds->register_callback_full(the_cib, a->update, 120, FALSE, strdup(a->id), "attrd_cib_callback", attrd_cib_callback, free); } free_xml(xml_top); } diff --git a/attrd/internal.h b/attrd/internal.h index 0207da85a0..0e854ac6b1 100644 --- a/attrd/internal.h +++ b/attrd/internal.h @@ -1,40 +1,40 @@ /* - * Copyright (C) 2013 Andrew Beekhof + * Copyright (C) 2013-2017 Andrew Beekhof * * This program 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 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 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 + cib_t *the_cib; -GMainLoop *mloop; -bool shutting_down; crm_cluster_t *attrd_cluster; GHashTable *attributes; election_t *writer; int attrd_error; #define attrd_send_ack(client, id, flags) \ crm_ipcs_send_ack((client), (id), (flags), "ack", __FUNCTION__, __LINE__) void write_attributes(bool all, bool peer_discovered); void attrd_peer_message(crm_node_t *client, xmlNode *msg); void attrd_client_peer_remove(const char *client_name, xmlNode *xml); void attrd_client_update(xmlNode *xml); void attrd_client_refresh(void); void attrd_client_query(crm_client_t *client, uint32_t id, uint32_t flags, xmlNode *query); void free_attribute(gpointer data); gboolean attrd_election_cb(gpointer user_data); void attrd_peer_change_cb(enum crm_status_type type, crm_node_t *peer, const void *data); diff --git a/attrd/legacy.c b/attrd/legacy.c index 126f54b599..666de658ee 100644 --- a/attrd/legacy.c +++ b/attrd/legacy.c @@ -1,1084 +1,998 @@ /* * Copyright (C) 2004 Andrew Beekhof * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU General Public * License as published by the Free Software Foundation; either * version 2 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 #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include -#include #include #include +#include + #define OPTARGS "hV" #if SUPPORT_HEARTBEAT ll_cluster_t *attrd_cluster_conn; #endif -GMainLoop *mainloop = NULL; char *attrd_uname = NULL; char *attrd_uuid = NULL; -gboolean need_shutdown = FALSE; GHashTable *attr_hash = NULL; cib_t *cib_conn = NULL; /* Convenience macro for registering a CIB callback. * Check cib_conn != NULL before using. */ #define register_cib_callback(call_id, data, fn, free_fn) \ cib_conn->cmds->register_callback_full(cib_conn, call_id, 120, FALSE, \ data, #fn, fn, free_fn) typedef struct attr_hash_entry_s { char *uuid; char *id; char *set; char *section; char *value; char *stored_value; int timeout; char *dampen; guint timer_id; char *user; } attr_hash_entry_t; void attrd_local_callback(xmlNode * msg); gboolean attrd_timer_callback(void *user_data); gboolean attrd_trigger_update(attr_hash_entry_t * hash_entry); void attrd_perform_update(attr_hash_entry_t * hash_entry); static void free_hash_entry(gpointer data) { attr_hash_entry_t *entry = data; if (entry == NULL) { return; } free(entry->id); free(entry->set); free(entry->dampen); free(entry->section); free(entry->uuid); free(entry->value); free(entry->stored_value); free(entry->user); free(entry); } -static int32_t -attrd_ipc_accept(qb_ipcs_connection_t * c, uid_t uid, gid_t gid) -{ - crm_trace("Connection %p", c); - if (need_shutdown) { - 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 -attrd_ipc_created(qb_ipcs_connection_t * c) -{ - crm_trace("Connection %p", c); -} - /* Exit code means? */ static int32_t attrd_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_ipcs_send_ack(client, id, flags, "ack", __FUNCTION__, __LINE__); if (msg == NULL) { crm_debug("No msg from %d (%p)", crm_ipcs_client_pid(c), c); return 0; } #if ENABLE_ACL CRM_ASSERT(client->user != NULL); crm_acl_get_set_user(msg, F_ATTRD_USER, client->user); #endif crm_trace("Processing msg from %d (%p)", crm_ipcs_client_pid(c), c); crm_log_xml_trace(msg, __FUNCTION__); attrd_local_callback(msg); free_xml(msg); return 0; } -/* Error code means? */ -static int32_t -attrd_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 -attrd_ipc_destroy(qb_ipcs_connection_t * c) -{ - crm_trace("Connection %p", c); - attrd_ipc_closed(c); -} - -struct qb_ipcs_service_handlers ipc_callbacks = { - .connection_accept = attrd_ipc_accept, - .connection_created = attrd_ipc_created, - .msg_process = attrd_ipc_dispatch, - .connection_closed = attrd_ipc_closed, - .connection_destroyed = attrd_ipc_destroy -}; - -static void -attrd_shutdown(int nsig) -{ - need_shutdown = TRUE; - crm_info("Exiting"); - if (mainloop != NULL && g_main_is_running(mainloop)) { - g_main_quit(mainloop); - } else { - crm_exit(pcmk_ok); - } -} - static void usage(const char *cmd, int exit_status) { FILE *stream; stream = exit_status ? stderr : stdout; fprintf(stream, "usage: %s [-srkh] [-c configure file]\n", cmd); /* fprintf(stream, "\t-d\tsets debug level\n"); */ /* fprintf(stream, "\t-s\tgets daemon status\n"); */ /* fprintf(stream, "\t-r\trestarts daemon\n"); */ /* fprintf(stream, "\t-k\tstops daemon\n"); */ /* fprintf(stream, "\t-h\thelp message\n"); */ fflush(stream); crm_exit(exit_status); } static void stop_attrd_timer(attr_hash_entry_t * hash_entry) { if (hash_entry != NULL && hash_entry->timer_id != 0) { crm_trace("Stopping %s timer", hash_entry->id); g_source_remove(hash_entry->timer_id); hash_entry->timer_id = 0; } } static void log_hash_entry(int level, attr_hash_entry_t * entry, const char *text) { do_crm_log(level, "%s: Set: %s, Name: %s, Value: %s, Timeout: %s", text, entry->section, entry->id, entry->value, entry->dampen); } static attr_hash_entry_t * find_hash_entry(xmlNode * msg) { const char *value = NULL; const char *attr = crm_element_value(msg, F_ATTRD_ATTRIBUTE); attr_hash_entry_t *hash_entry = NULL; if (attr == NULL) { crm_info("Ignoring message with no attribute name"); return NULL; } hash_entry = g_hash_table_lookup(attr_hash, attr); if (hash_entry == NULL) { /* create one and add it */ crm_info("Creating hash entry for %s", attr); hash_entry = calloc(1, sizeof(attr_hash_entry_t)); hash_entry->id = strdup(attr); g_hash_table_insert(attr_hash, hash_entry->id, hash_entry); hash_entry = g_hash_table_lookup(attr_hash, attr); CRM_CHECK(hash_entry != NULL, return NULL); } value = crm_element_value(msg, F_ATTRD_SET); if (value != NULL) { free(hash_entry->set); hash_entry->set = strdup(value); crm_debug("\t%s->set: %s", attr, value); } value = crm_element_value(msg, F_ATTRD_SECTION); if (value == NULL) { value = XML_CIB_TAG_STATUS; } free(hash_entry->section); hash_entry->section = strdup(value); crm_trace("\t%s->section: %s", attr, value); value = crm_element_value(msg, F_ATTRD_DAMPEN); if (value != NULL) { free(hash_entry->dampen); hash_entry->dampen = strdup(value); hash_entry->timeout = crm_get_msec(value); crm_trace("\t%s->timeout: %s", attr, value); } #if ENABLE_ACL free(hash_entry->user); hash_entry->user = NULL; value = crm_element_value(msg, F_ATTRD_USER); if (value != NULL) { hash_entry->user = strdup(value); crm_trace("\t%s->user: %s", attr, value); } #endif log_hash_entry(LOG_DEBUG_2, hash_entry, "Found (and updated) entry:"); return hash_entry; } static void process_xml_request(xmlNode *xml) { attr_hash_entry_t *hash_entry = NULL; const char *from = crm_element_value(xml, F_ORIG); const char *op = crm_element_value(xml, F_ATTRD_TASK); const char *host = crm_element_value(xml, F_ATTRD_HOST); const char *ignore = crm_element_value(xml, F_ATTRD_IGNORE_LOCALLY); if (host && safe_str_eq(host, attrd_uname)) { crm_info("Update relayed from %s", from); attrd_local_callback(xml); } else if (safe_str_eq(op, ATTRD_OP_PEER_REMOVE)) { CRM_CHECK(host != NULL, return); crm_debug("Removing %s from peer caches for %s", host, from); crm_remote_peer_cache_remove(host); reap_crm_member(0, host); } else if ((ignore == NULL) || safe_str_neq(from, attrd_uname)) { crm_trace("%s message from %s", op, from); hash_entry = find_hash_entry(xml); stop_attrd_timer(hash_entry); attrd_perform_update(hash_entry); } } #if SUPPORT_HEARTBEAT static void attrd_ha_connection_destroy(gpointer user_data) { crm_trace("Invoked"); - if (need_shutdown) { + if (attrd_shutting_down()) { /* we signed out, so this is expected */ crm_info("Heartbeat disconnection complete"); return; } crm_crit("Lost connection to heartbeat service!"); - if (mainloop != NULL && g_main_is_running(mainloop)) { - g_main_quit(mainloop); + if (attrd_mainloop_running()) { + attrd_quit_mainloop(); return; } crm_exit(pcmk_ok); } static void attrd_ha_callback(HA_Message * msg, void *private_data) { xmlNode *xml = convert_ha_message(NULL, msg, __FUNCTION__); process_xml_request(xml); free_xml(xml); } #endif #if SUPPORT_COROSYNC static void attrd_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("Bad message received: '%.120s'", data); } } if (xml != NULL) { /* crm_xml_add_int(xml, F_SEQ, wrapper->id); */ crm_xml_add(xml, F_ORIG, from); process_xml_request(xml); free_xml(xml); } free(data); } static void attrd_cs_destroy(gpointer unused) { - if (need_shutdown) { + if (attrd_shutting_down()) { /* we signed out, so this is expected */ crm_info("Corosync disconnection complete"); return; } crm_crit("Lost connection to Corosync service!"); - if (mainloop != NULL && g_main_is_running(mainloop)) { - g_main_quit(mainloop); + if (attrd_mainloop_running()) { + attrd_quit_mainloop(); return; } crm_exit(EINVAL); } #endif static void attrd_cib_connection_destroy(gpointer user_data) { cib_t *conn = user_data; conn->cmds->signoff(conn); /* Ensure IPC is cleaned up */ - if (need_shutdown) { + if (attrd_shutting_down()) { crm_info("Connection to the CIB terminated..."); } else { /* eventually this will trigger a reconnect, not a shutdown */ crm_err("Connection to the CIB terminated..."); crm_exit(ENOTCONN); } return; } static void update_for_hash_entry(gpointer key, gpointer value, gpointer user_data) { attr_hash_entry_t *entry = value; if (entry->value != NULL || entry->stored_value != NULL) { attrd_timer_callback(value); } } static void local_update_for_hash_entry(gpointer key, gpointer value, gpointer user_data) { attr_hash_entry_t *entry = value; if (entry->timer_id == 0) { crm_trace("Performing local-only update after replace for %s", entry->id); attrd_perform_update(entry); /* } else { * just let the timer expire and attrd_timer_callback() will do the right thing */ } } static void do_cib_replaced(const char *event, xmlNode * msg) { crm_info("Updating all attributes after %s event", event); g_hash_table_foreach(attr_hash, local_update_for_hash_entry, NULL); } static gboolean cib_connect(void *user_data) { static int attempts = 1; static int max_retry = 20; gboolean was_err = FALSE; static cib_t *local_conn = NULL; if (local_conn == NULL) { local_conn = cib_new(); } if (was_err == FALSE) { int rc = -ENOTCONN; if (attempts < max_retry) { crm_debug("CIB signon attempt %d", attempts); rc = local_conn->cmds->signon(local_conn, T_ATTRD, cib_command); } if (rc != pcmk_ok && attempts > max_retry) { crm_err("Signon to CIB failed: %s", pcmk_strerror(rc)); was_err = TRUE; } else if (rc != pcmk_ok) { attempts++; return TRUE; } } crm_info("Connected to the CIB after %d signon attempts", attempts); if (was_err == FALSE) { int rc = local_conn->cmds->set_connection_dnotify(local_conn, attrd_cib_connection_destroy); if (rc != pcmk_ok) { crm_err("Could not set dnotify callback"); was_err = TRUE; } } if (was_err == FALSE) { if (pcmk_ok != local_conn->cmds->add_notify_callback(local_conn, T_CIB_REPLACE_NOTIFY, do_cib_replaced)) { crm_err("Could not set CIB notification callback"); was_err = TRUE; } } if (was_err) { crm_err("Aborting startup"); crm_exit(DAEMON_RESPAWN_STOP); } cib_conn = local_conn; crm_info("Sending full refresh now that we're connected to the cib"); g_hash_table_foreach(attr_hash, local_update_for_hash_entry, NULL); return FALSE; } int main(int argc, char **argv) { int flag = 0; int argerr = 0; crm_cluster_t cluster; gboolean was_err = FALSE; qb_ipcs_connection_t *c = NULL; qb_ipcs_service_t *ipcs = NULL; crm_log_init(T_ATTRD, LOG_NOTICE, TRUE, FALSE, argc, argv, FALSE); mainloop_add_signal(SIGTERM, attrd_shutdown); while ((flag = getopt(argc, argv, OPTARGS)) != EOF) { switch (flag) { case 'V': crm_bump_log_level(argc, argv); break; case 'h': /* Help message */ usage(T_ATTRD, EX_OK); break; default: ++argerr; break; } } if (optind > argc) { ++argerr; } if (argerr) { usage(T_ATTRD, EX_USAGE); } attr_hash = g_hash_table_new_full(crm_str_hash, g_str_equal, NULL, free_hash_entry); crm_info("Starting up"); if (was_err == FALSE) { #if SUPPORT_COROSYNC if (is_openais_cluster()) { cluster.destroy = attrd_cs_destroy; cluster.cpg.cpg_deliver_fn = attrd_cs_dispatch; cluster.cpg.cpg_confchg_fn = pcmk_cpg_membership; } #endif #if SUPPORT_HEARTBEAT if (is_heartbeat_cluster()) { cluster.hb_conn = NULL; cluster.hb_dispatch = attrd_ha_callback; cluster.destroy = attrd_ha_connection_destroy; } #endif if (FALSE == crm_cluster_connect(&cluster)) { crm_err("HA Signon failed"); was_err = TRUE; } attrd_uname = cluster.uname; attrd_uuid = cluster.uuid; #if SUPPORT_HEARTBEAT attrd_cluster_conn = cluster.hb_conn; #endif } crm_info("Cluster connection active"); if (was_err == FALSE) { - attrd_ipc_server_init(&ipcs, &ipc_callbacks); + attrd_init_ipc(&ipcs, attrd_ipc_dispatch); } crm_info("Accepting attribute updates"); - mainloop = g_main_new(FALSE); + attrd_init_mainloop(); if (0 == g_timeout_add_full(G_PRIORITY_LOW + 1, 5000, cib_connect, NULL, NULL)) { crm_info("Adding timer failed"); was_err = TRUE; } if (was_err) { crm_err("Aborting startup"); return 100; } crm_notice("Starting mainloop..."); - g_main_run(mainloop); + attrd_run_mainloop(); crm_notice("Exiting..."); #if SUPPORT_HEARTBEAT if (is_heartbeat_cluster()) { attrd_cluster_conn->llc_ops->signoff(attrd_cluster_conn, TRUE); attrd_cluster_conn->llc_ops->delete(attrd_cluster_conn); } #endif c = qb_ipcs_connection_first_get(ipcs); while (c != NULL) { qb_ipcs_connection_t *last = c; c = qb_ipcs_connection_next_get(ipcs, last); /* There really shouldn't be anyone connected at this point */ crm_notice("Disconnecting client %p, pid=%d...", last, crm_ipcs_client_pid(last)); qb_ipcs_disconnect(last); qb_ipcs_connection_unref(last); } qb_ipcs_destroy(ipcs); if (cib_conn) { cib_conn->cmds->signoff(cib_conn); cib_delete(cib_conn); } g_hash_table_destroy(attr_hash); free(attrd_uuid); return crm_exit(pcmk_ok); } struct attrd_callback_s { char *attr; char *value; }; /*! * \internal * \brief Free an attrd callback structure */ static void free_attrd_callback(void *user_data) { struct attrd_callback_s *data = user_data; free(data->attr); free(data->value); free(data); } static void attrd_cib_callback(xmlNode * msg, int call_id, int rc, xmlNode * output, void *user_data) { attr_hash_entry_t *hash_entry = NULL; struct attrd_callback_s *data = user_data; if (data->value == NULL && rc == -ENXIO) { rc = pcmk_ok; } else if (call_id < 0) { crm_warn("Update %s=%s failed: %s", data->attr, data->value, pcmk_strerror(call_id)); return; } switch (rc) { case pcmk_ok: crm_debug("Update %d for %s=%s passed", call_id, data->attr, data->value); hash_entry = g_hash_table_lookup(attr_hash, data->attr); if (hash_entry) { free(hash_entry->stored_value); hash_entry->stored_value = NULL; if (data->value != NULL) { hash_entry->stored_value = strdup(data->value); } } 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 */ crm_warn("Update %d for %s=%s failed: %s", call_id, data->attr, data->value, pcmk_strerror(rc)); break; default: crm_err("Update %d for %s=%s failed: %s", call_id, data->attr, data->value, pcmk_strerror(rc)); } } void attrd_perform_update(attr_hash_entry_t * hash_entry) { int rc = pcmk_ok; struct attrd_callback_s *data = NULL; const char *user_name = NULL; if (hash_entry == NULL) { return; } else if (cib_conn == NULL) { crm_info("Delaying operation %s=%s: cib not connected", hash_entry->id, crm_str(hash_entry->value)); return; } #if ENABLE_ACL if (hash_entry->user) { user_name = hash_entry->user; crm_trace("Performing request from user '%s'", hash_entry->user); } #endif if (hash_entry->value == NULL) { /* delete the attr */ rc = delete_attr_delegate(cib_conn, cib_none, hash_entry->section, attrd_uuid, NULL, hash_entry->set, hash_entry->uuid, hash_entry->id, NULL, FALSE, user_name); if (rc >= 0 && hash_entry->stored_value) { crm_notice("Sent delete %d: node=%s, attr=%s, id=%s, set=%s, section=%s", rc, attrd_uuid, hash_entry->id, hash_entry->uuid ? hash_entry->uuid : "", hash_entry->set, hash_entry->section); } else if (rc < 0 && rc != -ENXIO) { crm_notice ("Delete operation failed: node=%s, attr=%s, id=%s, set=%s, section=%s: %s (%d)", attrd_uuid, hash_entry->id, hash_entry->uuid ? hash_entry->uuid : "", hash_entry->set, hash_entry->section, pcmk_strerror(rc), rc); } else { crm_trace("Sent delete %d: node=%s, attr=%s, id=%s, set=%s, section=%s", rc, attrd_uuid, hash_entry->id, hash_entry->uuid ? hash_entry->uuid : "", hash_entry->set, hash_entry->section); } } else { /* send update */ rc = update_attr_delegate(cib_conn, cib_none, hash_entry->section, attrd_uuid, NULL, hash_entry->set, hash_entry->uuid, hash_entry->id, hash_entry->value, FALSE, user_name, NULL); if (rc < 0) { crm_notice("Sent update %s=%s failed: %s", hash_entry->id, hash_entry->value, pcmk_strerror(rc)); } if (safe_str_neq(hash_entry->value, hash_entry->stored_value) || rc < 0) { crm_notice("Sent update %d: %s=%s", rc, hash_entry->id, hash_entry->value); } else { crm_trace("Sent update %d: %s=%s", rc, hash_entry->id, hash_entry->value); } } data = calloc(1, sizeof(struct attrd_callback_s)); data->attr = strdup(hash_entry->id); if (hash_entry->value != NULL) { data->value = strdup(hash_entry->value); } register_cib_callback(rc, data, attrd_cib_callback, free_attrd_callback); return; } -/* strlen("value") */ -#define plus_plus_len (5) - /*! * \internal * \brief Expand attribute values that use "++" or "+=" * * \param[in] value Attribute value to expand * \param[in] old_value Previous value of attribute * * \return Newly allocated string with expanded value, or NULL if not expanded */ static char * expand_attr_value(const char *value, const char *old_value) { - int value_len = strlen(value); char *expanded = NULL; - if ((value_len >= (plus_plus_len + 2)) - && (value[plus_plus_len] == '+') - && ((value[plus_plus_len + 1] == '+') - || (value[plus_plus_len + 1] == '='))) { - - int offset = 1; - int int_value = char2score(old_value); - - 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; - } - - expanded = crm_itoa(int_value); + if (attrd_value_needs_expansion(value)) { + expanded = crm_itoa(attrd_expand_value(value, old_value)); } return expanded; } /*! * \internal * \brief Update a single node attribute for this node * * \param[in] msg XML message with update * \param[in,out] hash_entry Node attribute structure */ static void update_local_attr(xmlNode *msg, attr_hash_entry_t *hash_entry) { const char *value = crm_element_value(msg, F_ATTRD_VALUE); char *expanded = NULL; if (hash_entry->uuid == NULL) { const char *key = crm_element_value(msg, F_ATTRD_KEY); if (key) { hash_entry->uuid = strdup(key); } } crm_debug("Supplied: %s, Current: %s, Stored: %s", value, hash_entry->value, hash_entry->stored_value); if (safe_str_eq(value, hash_entry->value) && safe_str_eq(value, hash_entry->stored_value)) { crm_trace("Ignoring non-change"); return; } else if (value) { expanded = expand_attr_value(value, hash_entry->value); if (expanded) { crm_info("Expanded %s=%s to %s", hash_entry->id, value, expanded); value = expanded; } } if (safe_str_eq(value, hash_entry->value) && hash_entry->timer_id) { /* We're already waiting to set this value */ free(expanded); return; } free(hash_entry->value); hash_entry->value = NULL; if (value != NULL) { hash_entry->value = (expanded? expanded : strdup(value)); crm_debug("New value of %s is %s", hash_entry->id, value); } stop_attrd_timer(hash_entry); if (hash_entry->timeout > 0) { hash_entry->timer_id = g_timeout_add(hash_entry->timeout, attrd_timer_callback, hash_entry); } else { attrd_trigger_update(hash_entry); } } /*! * \internal * \brief Log the result of a CIB operation for a remote attribute * * \param[in] msg ignored * \param[in] id CIB operation ID * \param[in] rc CIB operation result * \param[in] output ignored * \param[in] data User-friendly string describing operation */ static void remote_attr_callback(xmlNode *msg, int id, int rc, xmlNode *output, void *data) { if (rc == pcmk_ok) { crm_debug("%s succeeded " CRM_XS " call=%d", data, id); } else { crm_notice("%s failed: %s " CRM_XS " call=%d rc=%d", data, pcmk_strerror(rc), id, rc); } } /*! * \internal * \brief Update a Pacemaker Remote node attribute via CIB only * * \param[in] host Pacemaker Remote node name * \param[in] name Attribute name * \param[in] value New attribute value * \param[in] section CIB section to update (defaults to status if NULL) * \param[in] user_name User to perform operation as * * \note Legacy attrd does not track remote node attributes, so such requests * are only sent to the CIB. This means that dampening is ignored, and * updates for the same attribute submitted to different nodes cannot be * reliably ordered. This is not ideal, but allows remote nodes to * be supported, and should be acceptable in practice. */ static void update_remote_attr(const char *host, const char *name, const char *value, const char *section, const char *user_name) { int rc = pcmk_ok; char *desc; if (value == NULL) { desc = crm_strdup_printf("Delete of %s in %s for %s", name, section, host); } else { desc = crm_strdup_printf("Update of %s=%s in %s for %s", name, value, section, host); } if (name == NULL) { rc = -EINVAL; } else if (cib_conn == NULL) { rc = -ENOTCONN; } if (rc != pcmk_ok) { remote_attr_callback(NULL, rc, rc, NULL, desc); free(desc); return; } if (value == NULL) { rc = delete_attr_delegate(cib_conn, cib_none, section, host, NULL, NULL, NULL, name, NULL, FALSE, user_name); } else { rc = update_attr_delegate(cib_conn, cib_none, section, host, NULL, NULL, NULL, name, value, FALSE, user_name, "remote"); } crm_trace("%s submitted as CIB call %d", desc, rc); register_cib_callback(rc, desc, remote_attr_callback, free); } void attrd_local_callback(xmlNode * msg) { attr_hash_entry_t *hash_entry = NULL; const char *from = crm_element_value(msg, F_ORIG); const char *op = crm_element_value(msg, F_ATTRD_TASK); const char *attr = crm_element_value(msg, F_ATTRD_ATTRIBUTE); const char *pattern = crm_element_value(msg, F_ATTRD_REGEX); const char *value = crm_element_value(msg, F_ATTRD_VALUE); const char *host = crm_element_value(msg, F_ATTRD_HOST); int is_remote = FALSE; crm_element_value_int(msg, F_ATTRD_IS_REMOTE, &is_remote); if (safe_str_eq(op, ATTRD_OP_REFRESH)) { crm_notice("Sending full refresh (origin=%s)", from); g_hash_table_foreach(attr_hash, update_for_hash_entry, NULL); return; } else if (safe_str_eq(op, ATTRD_OP_PEER_REMOVE)) { if (host) { crm_notice("Broadcasting removal of peer %s", host); send_cluster_message(NULL, crm_msg_attrd, msg, FALSE); } return; } else if (op && safe_str_neq(op, ATTRD_OP_UPDATE)) { crm_notice("Ignoring unsupported %s request from %s", op, from); return; } /* Handle requests for Pacemaker Remote nodes specially */ if (host && is_remote) { const char *section = crm_element_value(msg, F_ATTRD_SECTION); const char *user_name = crm_element_value(msg, F_ATTRD_USER); if (section == NULL) { section = XML_CIB_TAG_STATUS; } if ((attr == NULL) && (pattern != NULL)) { /* Attribute(s) specified by regular expression */ /* @TODO query, iterate and update_remote_attr() for matches? */ crm_notice("Update of %s for %s failed: regular expressions " "are not supported with Pacemaker Remote nodes", pattern, host); } else { /* Single attribute specified by exact name */ update_remote_attr(host, attr, value, section, user_name); } return; } /* Redirect requests for another cluster node to that node */ if (host != NULL && safe_str_neq(host, attrd_uname)) { send_cluster_message(crm_get_peer(0, host), crm_msg_attrd, msg, FALSE); return; } if (attr != NULL) { /* Single attribute specified by exact name */ crm_debug("%s message from %s: %s=%s", op, from, attr, crm_str(value)); hash_entry = find_hash_entry(msg); if (hash_entry != NULL) { update_local_attr(msg, hash_entry); } } else if (pattern != NULL) { /* Attribute(s) specified by regular expression */ regex_t regex; GHashTableIter iter; if (regcomp(®ex, pattern, REG_EXTENDED|REG_NOSUB)) { crm_err("Update from %s failed: invalid pattern %s", from, pattern); return; } crm_debug("%s message from %s: %s=%s", op, from, pattern, crm_str(value)); g_hash_table_iter_init(&iter, attr_hash); while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &hash_entry)) { int rc = regexec(®ex, hash_entry->id, 0, NULL, 0); if (rc == 0) { crm_trace("Attribute %s matches %s", hash_entry->id, pattern); update_local_attr(msg, hash_entry); } } } else { crm_info("Ignoring message with no attribute name or expression"); } } gboolean attrd_timer_callback(void *user_data) { stop_attrd_timer(user_data); attrd_trigger_update(user_data); return TRUE; /* Always return true, removed cleanly by stop_attrd_timer() */ } gboolean attrd_trigger_update(attr_hash_entry_t * hash_entry) { xmlNode *msg = NULL; /* send HA message to everyone */ crm_notice("Sending flush op to all hosts for: %s (%s)", hash_entry->id, crm_str(hash_entry->value)); log_hash_entry(LOG_DEBUG_2, hash_entry, "Sending flush op to all hosts for:"); msg = create_xml_node(NULL, __FUNCTION__); crm_xml_add(msg, F_TYPE, T_ATTRD); crm_xml_add(msg, F_ORIG, attrd_uname); crm_xml_add(msg, F_ATTRD_TASK, "flush"); crm_xml_add(msg, F_ATTRD_ATTRIBUTE, hash_entry->id); crm_xml_add(msg, F_ATTRD_SET, hash_entry->set); crm_xml_add(msg, F_ATTRD_SECTION, hash_entry->section); crm_xml_add(msg, F_ATTRD_DAMPEN, hash_entry->dampen); crm_xml_add(msg, F_ATTRD_VALUE, hash_entry->value); #if ENABLE_ACL if (hash_entry->user) { crm_xml_add(msg, F_ATTRD_USER, hash_entry->user); } #endif if (hash_entry->timeout <= 0) { crm_xml_add(msg, F_ATTRD_IGNORE_LOCALLY, hash_entry->value); attrd_perform_update(hash_entry); } send_cluster_message(NULL, crm_msg_attrd, msg, FALSE); free_xml(msg); return TRUE; } diff --git a/attrd/main.c b/attrd/main.c index ccb24f3516..ec4a2eb033 100644 --- a/attrd/main.c +++ b/attrd/main.c @@ -1,399 +1,334 @@ /* * Copyright (C) 2013 Andrew Beekhof * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU General Public * License as published by the Free Software Foundation; either * version 2 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 #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include -#include #include #include #include cib_t *the_cib = NULL; -GMainLoop *mloop = NULL; -bool shutting_down = FALSE; crm_cluster_t *attrd_cluster = NULL; election_t *writer = NULL; int attrd_error = pcmk_ok; -static void -attrd_shutdown(int nsig) { - shutting_down = TRUE; - crm_info("Shutting down"); - - if (mloop != NULL && g_main_is_running(mloop)) { - g_main_quit(mloop); - } else { - crm_exit(pcmk_ok); - } -} - static void attrd_cpg_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("Bad message of class %d received from %s[%u]: '%.120s'", kind, from, nodeid, data); } else { crm_node_t *peer = crm_get_peer(nodeid, from); attrd_peer_message(peer, xml); } free_xml(xml); free(data); } static void attrd_cpg_destroy(gpointer unused) { - if (shutting_down) { + if (attrd_shutting_down()) { crm_info("Corosync disconnection complete"); } else { crm_crit("Lost connection to Corosync service!"); attrd_error = ECONNRESET; attrd_shutdown(0); } } static void attrd_cib_replaced_cb(const char *event, xmlNode * msg) { crm_notice("Updating all attributes after %s event", event); if(election_state(writer) == election_won) { write_attributes(TRUE, FALSE); } } static void attrd_cib_destroy_cb(gpointer user_data) { cib_t *conn = user_data; conn->cmds->signoff(conn); /* Ensure IPC is cleaned up */ - if (shutting_down) { + if (attrd_shutting_down()) { crm_info("Connection disconnection complete"); } else { /* eventually this should trigger a reconnect, not a shutdown */ crm_err("Lost connection to CIB service!"); attrd_error = ECONNRESET; attrd_shutdown(0); } return; } static cib_t * attrd_cib_connect(int max_retry) { int rc = -ENOTCONN; static int attempts = 0; cib_t *connection = cib_new(); if(connection == NULL) { return NULL; } do { if(attempts > 0) { sleep(attempts); } attempts++; crm_debug("CIB signon attempt %d", attempts); rc = connection->cmds->signon(connection, T_ATTRD, cib_command); } while(rc != pcmk_ok && attempts < max_retry); if (rc != pcmk_ok) { crm_err("Signon to CIB failed: %s (%d)", pcmk_strerror(rc), rc); goto cleanup; } crm_info("Connected to the CIB after %d attempts", attempts); rc = connection->cmds->set_connection_dnotify(connection, attrd_cib_destroy_cb); if (rc != pcmk_ok) { crm_err("Could not set disconnection callback"); goto cleanup; } rc = connection->cmds->add_notify_callback(connection, T_CIB_REPLACE_NOTIFY, attrd_cib_replaced_cb); if(rc != pcmk_ok) { crm_err("Could not set CIB notification callback"); goto cleanup; } return connection; cleanup: connection->cmds->signoff(connection); cib_delete(connection); return NULL; } -static int32_t -attrd_ipc_accept(qb_ipcs_connection_t * c, uid_t uid, gid_t gid) -{ - crm_trace("Connection %p", c); - if (shutting_down) { - 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 -attrd_ipc_created(qb_ipcs_connection_t * c) -{ - crm_trace("Connection %p", c); -} - static int32_t attrd_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 *xml = crm_ipcs_recv(client, data, size, &id, &flags); const char *op; if (xml == NULL) { crm_debug("No msg from %d (%p)", crm_ipcs_client_pid(c), c); return 0; } #if ENABLE_ACL CRM_ASSERT(client->user != NULL); crm_acl_get_set_user(xml, F_ATTRD_USER, client->user); #endif crm_trace("Processing msg from %d (%p)", crm_ipcs_client_pid(c), c); crm_log_xml_trace(xml, __FUNCTION__); op = crm_element_value(xml, F_ATTRD_TASK); if (client->name == NULL) { const char *value = crm_element_value(xml, F_ORIG); client->name = crm_strdup_printf("%s.%d", value?value:"unknown", client->pid); } if (safe_str_eq(op, ATTRD_OP_PEER_REMOVE)) { attrd_send_ack(client, id, flags); attrd_client_peer_remove(client->name, xml); } else if (safe_str_eq(op, ATTRD_OP_UPDATE)) { attrd_send_ack(client, id, flags); attrd_client_update(xml); } else if (safe_str_eq(op, ATTRD_OP_UPDATE_BOTH)) { attrd_send_ack(client, id, flags); attrd_client_update(xml); } else if (safe_str_eq(op, ATTRD_OP_UPDATE_DELAY)) { attrd_send_ack(client, id, flags); attrd_client_update(xml); } else if (safe_str_eq(op, ATTRD_OP_REFRESH)) { attrd_send_ack(client, id, flags); attrd_client_refresh(); } else if (safe_str_eq(op, ATTRD_OP_QUERY)) { /* queries will get reply, so no ack is necessary */ attrd_client_query(client, id, flags, xml); } else { crm_info("Ignoring request from client %s with unknown operation %s", client->name, op); } free_xml(xml); return 0; } -/* Error code means? */ -static int32_t -attrd_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 -attrd_ipc_destroy(qb_ipcs_connection_t * c) -{ - crm_trace("Connection %p", c); - attrd_ipc_closed(c); -} - -struct qb_ipcs_service_handlers ipc_callbacks = { - .connection_accept = attrd_ipc_accept, - .connection_created = attrd_ipc_created, - .msg_process = attrd_ipc_dispatch, - .connection_closed = attrd_ipc_closed, - .connection_destroyed = attrd_ipc_destroy -}; - /* *INDENT-OFF* */ static struct crm_option long_options[] = { /* Top-level Options */ {"help", 0, 0, '?', "\tThis text"}, {"verbose", 0, 0, 'V', "\tIncrease debug output"}, {0, 0, 0, 0} }; /* *INDENT-ON* */ int main(int argc, char **argv) { int rc = pcmk_ok; int flag = 0; int index = 0; int argerr = 0; qb_ipcs_service_t *ipcs = NULL; - mloop = g_main_new(FALSE); + attrd_init_mainloop(); crm_log_preinit(NULL, argc, argv); crm_set_options(NULL, "[options]", long_options, "Daemon for aggregating and atomically storing node attribute updates into the CIB"); mainloop_add_signal(SIGTERM, attrd_shutdown); 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 'h': /* Help message */ crm_help(flag, EX_OK); break; default: ++argerr; break; } } if (optind > argc) { ++argerr; } if (argerr) { crm_help('?', EX_USAGE); } crm_log_init(T_ATTRD, LOG_INFO, TRUE, FALSE, argc, argv, FALSE); crm_info("Starting up"); attributes = g_hash_table_new_full(crm_str_hash, g_str_equal, NULL, free_attribute); attrd_cluster = malloc(sizeof(crm_cluster_t)); attrd_cluster->destroy = attrd_cpg_destroy; attrd_cluster->cpg.cpg_deliver_fn = attrd_cpg_dispatch; attrd_cluster->cpg.cpg_confchg_fn = pcmk_cpg_membership; crm_set_status_callback(&attrd_peer_change_cb); if (crm_cluster_connect(attrd_cluster) == FALSE) { crm_err("Cluster connection failed"); rc = DAEMON_RESPAWN_STOP; goto done; } crm_info("Cluster connection active"); writer = election_init(T_ATTRD, attrd_cluster->uname, 120000, attrd_election_cb); - attrd_ipc_server_init(&ipcs, &ipc_callbacks); + attrd_init_ipc(&ipcs, attrd_ipc_dispatch); crm_info("Accepting attribute updates"); the_cib = attrd_cib_connect(10); if (the_cib == NULL) { rc = DAEMON_RESPAWN_STOP; goto done; } crm_info("CIB connection active"); - g_main_run(mloop); + attrd_run_mainloop(); done: crm_info("Shutting down attribute manager"); election_fini(writer); if (ipcs) { crm_client_disconnect_all(ipcs); qb_ipcs_destroy(ipcs); g_hash_table_destroy(attributes); } if (the_cib) { the_cib->cmds->signoff(the_cib); cib_delete(the_cib); } if(attrd_error) { return crm_exit(attrd_error); } return crm_exit(rc); }