diff --git a/daemons/attrd/attrd_ipc.c b/daemons/attrd/attrd_ipc.c index 845f0fcb61..6b9a6b41bf 100644 --- a/daemons/attrd/attrd_ipc.c +++ b/daemons/attrd/attrd_ipc.c @@ -1,532 +1,532 @@ /* * Copyright 2004-2022 the Pacemaker project contributors * * The version control history for this file may have further details. * * 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 #include #include #include #include #include #include "pacemaker-attrd.h" static qb_ipcs_service_t *ipcs = NULL; /*! * \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, __func__); attribute_t *a; if (reply == NULL) { return NULL; } crm_xml_add(reply, F_TYPE, T_ATTRD); crm_xml_add(reply, F_SUBTYPE, PCMK__ATTRD_CMD_QUERY); crm_xml_add(reply, PCMK__XA_ATTR_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, PCMK__XA_ATTR_NAME, attr); /* Allow caller to use "localhost" to refer to local node */ if (pcmk__str_eq(host, "localhost", pcmk__str_casei)) { 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; } pcmk__xe_add_node(host_value, host, 0); crm_xml_add(host_value, PCMK__XA_ATTR_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; } pcmk__xe_add_node(host_value, v->nodename, 0); crm_xml_add(host_value, PCMK__XA_ATTR_VALUE, v->current); } } } return reply; } xmlNode * attrd_client_clear_failure(pcmk__request_t *request) { xmlNode *xml = request->xml; const char *rsc, *op, *interval_spec; if (minimum_protocol_version >= 2) { /* Propagate to all peers (including ourselves). * This ends up at attrd_peer_message(). */ attrd_send_message(NULL, xml, false); pcmk__set_result(&request->result, CRM_EX_OK, PCMK_EXEC_DONE, NULL); return NULL; } rsc = crm_element_value(xml, PCMK__XA_ATTR_RESOURCE); op = crm_element_value(xml, PCMK__XA_ATTR_OPERATION); interval_spec = crm_element_value(xml, PCMK__XA_ATTR_INTERVAL); /* Map this to an update */ crm_xml_add(xml, PCMK__XA_TASK, PCMK__ATTRD_CMD_UPDATE); /* Add regular expression matching desired attributes */ if (rsc) { char *pattern; if (op == NULL) { pattern = crm_strdup_printf(ATTRD_RE_CLEAR_ONE, rsc); } else { guint interval_ms = crm_parse_interval_spec(interval_spec); pattern = crm_strdup_printf(ATTRD_RE_CLEAR_OP, rsc, op, interval_ms); } crm_xml_add(xml, PCMK__XA_ATTR_PATTERN, pattern); free(pattern); } else { crm_xml_add(xml, PCMK__XA_ATTR_PATTERN, ATTRD_RE_CLEAR_ALL); } /* Make sure attribute and value are not set, so we delete via regex */ if (crm_element_value(xml, PCMK__XA_ATTR_NAME)) { crm_xml_replace(xml, PCMK__XA_ATTR_NAME, NULL); } if (crm_element_value(xml, PCMK__XA_ATTR_VALUE)) { crm_xml_replace(xml, PCMK__XA_ATTR_VALUE, NULL); } return attrd_client_update(request); } xmlNode * attrd_client_peer_remove(pcmk__request_t *request) { xmlNode *xml = request->xml; // Host and ID are not used in combination, rather host has precedence const char *host = crm_element_value(xml, PCMK__XA_ATTR_NODE_NAME); char *host_alloc = NULL; attrd_send_ack(request->ipc_client, request->ipc_id, request->ipc_flags); if (host == NULL) { int nodeid = 0; crm_element_value_int(xml, PCMK__XA_ATTR_NODE_ID, &nodeid); if (nodeid > 0) { crm_node_t *node = pcmk__search_cluster_node_cache(nodeid, NULL); char *host_alloc = NULL; if (node && node->uname) { // Use cached name if available host = node->uname; } else { // Otherwise ask cluster layer host_alloc = get_node_name(nodeid); host = host_alloc; } pcmk__xe_add_node(xml, host, 0); } } if (host) { crm_info("Client %s is requesting all values for %s be removed", pcmk__client_name(request->ipc_client), host); attrd_send_message(NULL, xml, false); /* ends up at attrd_peer_message() */ free(host_alloc); } else { crm_info("Ignoring request by client %s to remove all peer values without specifying peer", pcmk__client_name(request->ipc_client)); } pcmk__set_result(&request->result, CRM_EX_OK, PCMK_EXEC_DONE, NULL); return NULL; } xmlNode * attrd_client_query(pcmk__request_t *request) { xmlNode *query = request->xml; xmlNode *reply = NULL; const char *attr = NULL; crm_debug("Query arrived from %s", pcmk__client_name(request->ipc_client)); /* Request must specify attribute name to query */ attr = crm_element_value(query, PCMK__XA_ATTR_NAME); if (attr == NULL) { pcmk__format_result(&request->result, CRM_EX_ERROR, PCMK_EXEC_ERROR, "Ignoring malformed query from %s (no attribute name given)", pcmk__client_name(request->ipc_client)); return NULL; } /* Build the XML reply */ reply = build_query_reply(attr, crm_element_value(query, PCMK__XA_ATTR_NODE_NAME)); if (reply == NULL) { pcmk__format_result(&request->result, CRM_EX_ERROR, PCMK_EXEC_ERROR, "Could not respond to query from %s: could not create XML reply", pcmk__client_name(request->ipc_client)); return NULL; } else { pcmk__set_result(&request->result, CRM_EX_OK, PCMK_EXEC_DONE, NULL); } request->ipc_client->request_id = 0; return reply; } xmlNode * attrd_client_refresh(pcmk__request_t *request) { crm_info("Updating all attributes"); attrd_send_ack(request->ipc_client, request->ipc_id, request->ipc_flags); attrd_write_attributes(true, true); pcmk__set_result(&request->result, CRM_EX_OK, PCMK_EXEC_DONE, NULL); return NULL; } xmlNode * attrd_client_update(pcmk__request_t *request) { xmlNode *xml = request->xml; attribute_t *a = NULL; char *host; const char *attr, *value, *regex; /* If the message has children, that means it is a message from a newer * client that supports sending multiple operations at a time. There are * two ways we can handle that. */ if (xml_has_children(xml)) { - if (minimum_protocol_version >= 4) { + if (ATTRD_SUPPORTS_MULTI_MESSAGE(minimum_protocol_version)) { /* First, if all peers support a certain protocol version, we can * just broadcast the big message and they'll handle it. */ attrd_send_message(NULL, xml, false); pcmk__set_result(&request->result, CRM_EX_OK, PCMK_EXEC_DONE, NULL); } else { /* Second, if they do not support that protocol version, split it * up into individual messages and call attrd_client_update on * each one. */ for (xmlNode *child = first_named_child(xml, XML_ATTR_OP); child != NULL; child = crm_next_same_xml(child)) { request->xml = child; /* Calling pcmk__set_result is handled by one of these calls to * attrd_client_update, so no need to do it again here. */ attrd_client_update(request); } } return NULL; } host = crm_element_value_copy(xml, PCMK__XA_ATTR_NODE_NAME); attr = crm_element_value(xml, PCMK__XA_ATTR_NAME); value = crm_element_value(xml, PCMK__XA_ATTR_VALUE); regex = crm_element_value(xml, PCMK__XA_ATTR_PATTERN); /* 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|REG_NOSUB)) { pcmk__format_result(&request->result, CRM_EX_ERROR, PCMK_EXEC_ERROR, "Bad regex '%s' for update from client %s", regex, pcmk__client_name(request->ipc_client)); } 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, PCMK__XA_ATTR_NAME, attr); attrd_send_message(NULL, xml, false); } } pcmk__set_result(&request->result, CRM_EX_OK, PCMK_EXEC_DONE, NULL); } free(host); regfree(r_patt); free(r_patt); return NULL; } else if (attr == NULL) { crm_err("Update request did not specify attribute or regular expression"); pcmk__format_result(&request->result, CRM_EX_ERROR, PCMK_EXEC_ERROR, "Client %s update request did not specify attribute or regular expression", pcmk__client_name(request->ipc_client)); free(host); return NULL; } if (host == NULL) { crm_trace("Inferring host"); host = strdup(attrd_cluster->uname); pcmk__xe_add_node(xml, host, attrd_cluster->nodeid); } a = g_hash_table_lookup(attributes, attr); if (a == NULL && pcmk__str_eq(request->op, PCMK__ATTRD_CMD_UPDATE_DELAY, pcmk__str_none)) { pcmk__format_result(&request->result, CRM_EX_NOSUCH, PCMK_EXEC_ERROR, "Attribute %s does not exist", attr); free(host); return NULL; } /* If value was specified using ++ or += notation, expand to real value */ if (value) { if (attrd_value_needs_expansion(value)) { int int_value; attribute_value_t *v = NULL; if (a) { v = g_hash_table_lookup(a->values, host); } 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, PCMK__XA_ATTR_VALUE, int_value); /* Replacing the value frees the previous memory, so re-query it */ value = crm_element_value(xml, PCMK__XA_ATTR_VALUE); } } crm_debug("Broadcasting %s[%s]=%s%s", attr, host, value, (attrd_election_won()? " (writer)" : "")); free(host); if (pcmk__str_eq(attrd_request_sync_point(xml), PCMK__VALUE_CLUSTER, pcmk__str_none)) { /* The client is waiting on the cluster-wide sync point. In this case, * the response ACK is not sent until this attrd broadcasts the update * and receives its own confirmation back from all peers. */ attrd_expect_confirmations(request, attrd_cluster_sync_point_update); attrd_send_message(NULL, xml, true); /* ends up at attrd_peer_message() */ } else { /* The client is either waiting on the local sync point or was not * waiting on any sync point at all. For the local sync point, the * response ACK is sent in attrd_peer_update. For clients not * waiting on any sync point, the response ACK is sent in * handle_update_request immediately before this function was called. */ attrd_send_message(NULL, xml, false); /* ends up at attrd_peer_message() */ } pcmk__set_result(&request->result, CRM_EX_OK, PCMK_EXEC_DONE, NULL); return NULL; } /*! * \internal * \brief Accept a new client IPC connection * * \param[in,out] 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 (attrd_shutting_down()) { crm_info("Ignoring new connection from pid %d during shutdown", pcmk__client_pid(c)); return -EPERM; } if (pcmk__new_client(c, uid, gid) == NULL) { return -EIO; } return pcmk_ok; } /*! * \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) { pcmk__client_t *client = pcmk__find_client(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); /* Remove the client from the sync point waitlist if it's present. */ attrd_remove_client_from_waitlist(client); /* And no longer wait for confirmations from any peers. */ attrd_do_not_wait_for_client(client); pcmk__free_client(client); } return FALSE; } /*! * \internal * \brief Destroy a client IPC connection * * \param[in,out] 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); } static int32_t attrd_ipc_dispatch(qb_ipcs_connection_t * c, void *data, size_t size) { uint32_t id = 0; uint32_t flags = 0; pcmk__client_t *client = pcmk__find_client(c); xmlNode *xml = NULL; // Sanity-check, and parse XML from IPC data CRM_CHECK((c != NULL) && (client != NULL), return 0); if (data == NULL) { crm_debug("No IPC data from PID %d", pcmk__client_pid(c)); return 0; } xml = pcmk__client_data2xml(client, data, &id, &flags); if (xml == NULL) { crm_debug("Unrecognizable IPC data from PID %d", pcmk__client_pid(c)); pcmk__ipc_send_ack(client, id, flags, "ack", NULL, CRM_EX_PROTOCOL); return 0; } else { pcmk__request_t request = { .ipc_client = client, .ipc_id = id, .ipc_flags = flags, .peer = NULL, .xml = xml, .call_options = 0, .result = PCMK__UNKNOWN_RESULT, }; CRM_ASSERT(client->user != NULL); pcmk__update_acl_user(xml, PCMK__XA_ATTR_USER, client->user); request.op = crm_element_value_copy(request.xml, PCMK__XA_TASK); CRM_CHECK(request.op != NULL, return 0); attrd_handle_request(&request); pcmk__reset_request(&request); } free_xml(xml); return 0; } static struct qb_ipcs_service_handlers ipc_callbacks = { .connection_accept = attrd_ipc_accept, .connection_created = NULL, .msg_process = attrd_ipc_dispatch, .connection_closed = attrd_ipc_closed, .connection_destroyed = attrd_ipc_destroy }; void attrd_ipc_fini(void) { if (ipcs != NULL) { pcmk__drop_all_clients(ipcs); qb_ipcs_destroy(ipcs); ipcs = NULL; } } /*! * \internal * \brief Set up attrd IPC communication */ void attrd_init_ipc(void) { pcmk__serve_attrd_ipc(&ipcs, &ipc_callbacks); } diff --git a/daemons/attrd/attrd_sync.c b/daemons/attrd/attrd_sync.c index d3d7108a37..e48f82e691 100644 --- a/daemons/attrd/attrd_sync.c +++ b/daemons/attrd/attrd_sync.c @@ -1,522 +1,522 @@ /* * Copyright 2022 the Pacemaker project contributors * * The version control history for this file may have further details. * * This source code is licensed under the GNU General Public License version 2 * or later (GPLv2+) WITHOUT ANY WARRANTY. */ #include #include #include #include "pacemaker-attrd.h" /* A hash table storing clients that are waiting on a sync point to be reached. * The key is waitlist_client - just a plain int. The obvious key would be * the IPC client's ID, but this is not guaranteed to be unique. A single client * could be waiting on a sync point for multiple attributes at the same time. * * It is not expected that this hash table will ever be especially large. */ static GHashTable *waitlist = NULL; static int waitlist_client = 0; struct waitlist_node { /* What kind of sync point does this node describe? */ enum attrd_sync_point sync_point; /* Information required to construct and send a reply to the client. */ char *client_id; uint32_t ipc_id; uint32_t flags; }; /* A hash table storing information on in-progress IPC requests that are awaiting * confirmations. These requests are currently being processed by peer attrds and * we are waiting to receive confirmation messages from each peer indicating that * processing is complete. * * Multiple requests could be waiting on confirmations at the same time. * * The key is the unique callid for the IPC request, and the value is a * confirmation_action struct. */ static GHashTable *expected_confirmations = NULL; /*! * \internal * \brief A structure describing a single IPC request that is awaiting confirmations */ struct confirmation_action { /*! * \brief A list of peer attrds that we are waiting to receive confirmation * messages from * * This list is dynamic - as confirmations arrive from peer attrds, they will * be removed from this list. When the list is empty, all peers have processed * the request and the associated confirmation action will be taken. */ GList *respondents; /*! * \brief A function to run when all confirmations have been received */ attrd_confirmation_action_fn fn; /*! * \brief Information required to construct and send a reply to the client */ char *client_id; uint32_t ipc_id; uint32_t flags; /*! * \brief The XML request containing the callid associated with this action */ void *xml; }; static void next_key(void) { do { waitlist_client++; if (waitlist_client < 0) { waitlist_client = 1; } } while (g_hash_table_contains(waitlist, GINT_TO_POINTER(waitlist_client))); } static void free_waitlist_node(gpointer data) { struct waitlist_node *wl = (struct waitlist_node *) data; free(wl->client_id); free(wl); } static const char * sync_point_str(enum attrd_sync_point sync_point) { if (sync_point == attrd_sync_point_local) { return PCMK__VALUE_LOCAL; } else if (sync_point == attrd_sync_point_cluster) { return PCMK__VALUE_CLUSTER; } else { return "unknown"; } } /*! * \internal * \brief Add a client to the attrd waitlist * * Typically, a client receives an ACK for its XML IPC request immediately. However, * some clients want to wait until their request has been processed and taken effect. * This is called a sync point. Any client placed on this waitlist will have its * ACK message delayed until either its requested sync point is hit, or until it * times out. * * The XML IPC request must specify the type of sync point it wants to wait for. * * \param[in,out] request The request describing the client to place on the waitlist. */ void attrd_add_client_to_waitlist(pcmk__request_t *request) { const char *sync_point = attrd_request_sync_point(request->xml); struct waitlist_node *wl = NULL; if (sync_point == NULL) { return; } if (waitlist == NULL) { waitlist = pcmk__intkey_table(free_waitlist_node); } wl = calloc(sizeof(struct waitlist_node), 1); CRM_ASSERT(wl != NULL); wl->client_id = strdup(request->ipc_client->id); CRM_ASSERT(wl->client_id); if (pcmk__str_eq(sync_point, PCMK__VALUE_LOCAL, pcmk__str_none)) { wl->sync_point = attrd_sync_point_local; } else if (pcmk__str_eq(sync_point, PCMK__VALUE_CLUSTER, pcmk__str_none)) { wl->sync_point = attrd_sync_point_cluster; } else { free_waitlist_node(wl); return; } wl->ipc_id = request->ipc_id; wl->flags = request->flags; next_key(); pcmk__intkey_table_insert(waitlist, waitlist_client, wl); crm_trace("Added client %s to waitlist for %s sync point", wl->client_id, sync_point_str(wl->sync_point)); crm_trace("%d clients now on waitlist", g_hash_table_size(waitlist)); /* And then add the key to the request XML so we can uniquely identify * it when it comes time to issue the ACK. */ crm_xml_add_int(request->xml, XML_LRM_ATTR_CALLID, waitlist_client); } /*! * \internal * \brief Free all memory associated with the waitlist. This is most typically * used when attrd shuts down. */ void attrd_free_waitlist(void) { if (waitlist == NULL) { return; } g_hash_table_destroy(waitlist); waitlist = NULL; } /*! * \internal * \brief Unconditionally remove a client from the waitlist, such as when the client * node disconnects from the cluster * * \param[in] client The client to remove */ void attrd_remove_client_from_waitlist(pcmk__client_t *client) { GHashTableIter iter; gpointer value; if (waitlist == NULL) { return; } g_hash_table_iter_init(&iter, waitlist); while (g_hash_table_iter_next(&iter, NULL, &value)) { struct waitlist_node *wl = (struct waitlist_node *) value; if (wl->client_id == client->id) { g_hash_table_iter_remove(&iter); crm_trace("%d clients now on waitlist", g_hash_table_size(waitlist)); } } } /*! * \internal * \brief Send an IPC ACK message to all awaiting clients * * This function will search the waitlist for all clients that are currently awaiting * an ACK indicating their attrd operation is complete. Only those clients with a * matching sync point type and callid from their original XML IPC request will be * ACKed. Once they have received an ACK, they will be removed from the waitlist. * * \param[in] sync_point What kind of sync point have we hit? * \param[in] xml The original XML IPC request. */ void attrd_ack_waitlist_clients(enum attrd_sync_point sync_point, const xmlNode *xml) { int callid; gpointer value; if (waitlist == NULL) { return; } if (crm_element_value_int(xml, XML_LRM_ATTR_CALLID, &callid) == -1) { crm_warn("Could not get callid from request XML"); return; } value = pcmk__intkey_table_lookup(waitlist, callid); if (value != NULL) { struct waitlist_node *wl = (struct waitlist_node *) value; pcmk__client_t *client = NULL; if (wl->sync_point != sync_point) { return; } crm_trace("Alerting client %s for reached %s sync point", wl->client_id, sync_point_str(wl->sync_point)); client = pcmk__find_client_by_id(wl->client_id); if (client == NULL) { return; } attrd_send_ack(client, wl->ipc_id, wl->flags | crm_ipc_client_response); /* And then remove the client so it doesn't get alerted again. */ pcmk__intkey_table_remove(waitlist, callid); crm_trace("%d clients now on waitlist", g_hash_table_size(waitlist)); } } /*! * \internal * \brief Action to take when a cluster sync point is hit for a * PCMK__ATTRD_CMD_UPDATE* message. * * \param[in] xml The request that should be passed along to * attrd_ack_waitlist_clients. This should be the original * IPC request containing the callid for this update message. */ int attrd_cluster_sync_point_update(xmlNode *xml) { crm_trace("Hit cluster sync point for attribute update"); attrd_ack_waitlist_clients(attrd_sync_point_cluster, xml); return pcmk_rc_ok; } /*! * \internal * \brief Return the sync point attribute for an IPC request * * This function will check both the top-level element of \p xml for a sync * point attribute, as well as all of its \p op children, if any. The latter * is useful for newer versions of attrd that can put multiple IPC requests * into a single message. * * \param[in] xml An XML IPC request * * \note It is assumed that if one child element has a sync point attribute, * all will have a sync point attribute and they will all be the same * sync point. No other configuration is supported. * * \return The sync point attribute of \p xml, or NULL if none. */ const char * attrd_request_sync_point(xmlNode *xml) { if (xml_has_children(xml)) { xmlNode *child = pcmk__xe_match(xml, XML_ATTR_OP, PCMK__XA_ATTR_SYNC_POINT, NULL); if (child) { return crm_element_value(child, PCMK__XA_ATTR_SYNC_POINT); } else { return NULL; } } else { return crm_element_value(xml, PCMK__XA_ATTR_SYNC_POINT); } } /*! * \internal * \brief Does an IPC request contain any sync point attribute? * * \param[in] xml An XML IPC request * * \return true if there's a sync point attribute, false otherwise */ bool attrd_request_has_sync_point(xmlNode *xml) { return attrd_request_sync_point(xml) != NULL; } static void free_action(gpointer data) { struct confirmation_action *action = (struct confirmation_action *) data; g_list_free_full(action->respondents, free); free_xml(action->xml); free(action->client_id); free(action); } /*! * \internal * \brief When a peer disconnects from the cluster, no longer wait for its confirmation * for any IPC action. If this peer is the last one being waited on, this will * trigger the confirmation action. * * \param[in] host The disconnecting peer attrd's uname */ void attrd_do_not_expect_from_peer(const char *host) { GList *keys = g_hash_table_get_keys(expected_confirmations); crm_trace("Removing peer %s from expected confirmations", host); for (GList *node = keys; node != NULL; node = node->next) { int callid = *(int *) node->data; attrd_handle_confirmation(callid, host); } g_list_free(keys); } /*! * \internal * \brief When a client disconnects from the cluster, no longer wait on confirmations * for it. Because the peer attrds may still be processing the original IPC * message, they may still send us confirmations. However, we will take no * action on them. * * \param[in] client The disconnecting client */ void attrd_do_not_wait_for_client(pcmk__client_t *client) { GHashTableIter iter; gpointer value; if (expected_confirmations == NULL) { return; } g_hash_table_iter_init(&iter, expected_confirmations); while (g_hash_table_iter_next(&iter, NULL, &value)) { struct confirmation_action *action = (struct confirmation_action *) value; if (pcmk__str_eq(action->client_id, client->id, pcmk__str_none)) { crm_trace("Removing client %s from expected confirmations", client->id); g_hash_table_iter_remove(&iter); crm_trace("%d requests now in expected confirmations table", g_hash_table_size(expected_confirmations)); break; } } } /*! * \internal * \brief Register some action to be taken when IPC request confirmations are * received * * When this function is called, a list of all peer attrds that support confirming * requests is generated. As confirmations from these peer attrds are received, * they are removed from this list. When the list is empty, the registered action * will be called. * * \note This function should always be called before attrd_send_message is called * to broadcast to the peers to ensure that we know what replies we are * waiting on. Otherwise, it is possible the peer could finish and confirm * before we know to expect it. * * \param[in] request The request that is awaiting confirmations * \param[in] fn A function to be run after all confirmations are received */ void attrd_expect_confirmations(pcmk__request_t *request, attrd_confirmation_action_fn fn) { struct confirmation_action *action = NULL; GHashTableIter iter; gpointer host, ver; GList *respondents = NULL; int callid; if (expected_confirmations == NULL) { expected_confirmations = pcmk__intkey_table((GDestroyNotify) free_action); } if (crm_element_value_int(request->xml, XML_LRM_ATTR_CALLID, &callid) == -1) { crm_err("Could not get callid from xml"); return; } if (pcmk__intkey_table_lookup(expected_confirmations, callid)) { crm_err("Already waiting on confirmations for call id %d", callid); return; } g_hash_table_iter_init(&iter, peer_protocol_vers); while (g_hash_table_iter_next(&iter, &host, &ver)) { - if (GPOINTER_TO_INT(ver) >= 5) { + if (ATTRD_SUPPORTS_CONFIRMATION(GPOINTER_TO_INT(ver))) { char *s = strdup((char *) host); CRM_ASSERT(s != NULL); respondents = g_list_prepend(respondents, s); } } action = calloc(1, sizeof(struct confirmation_action)); CRM_ASSERT(action != NULL); action->respondents = respondents; action->fn = fn; action->xml = copy_xml(request->xml); action->client_id = strdup(request->ipc_client->id); CRM_ASSERT(action->client_id != NULL); action->ipc_id = request->ipc_id; action->flags = request->flags; pcmk__intkey_table_insert(expected_confirmations, callid, action); crm_trace("Callid %d now waiting on %d confirmations", callid, g_list_length(respondents)); crm_trace("%d requests now in expected confirmations table", g_hash_table_size(expected_confirmations)); } void attrd_free_confirmations(void) { if (expected_confirmations != NULL) { g_hash_table_destroy(expected_confirmations); expected_confirmations = NULL; } } /*! * \internal * \brief Process a confirmation message from a peer attrd * * This function is called every time a PCMK__ATTRD_CMD_CONFIRM message is * received from a peer attrd. If this is the last confirmation we are waiting * on for a given operation, the registered action will be called. * * \param[in] callid The unique callid for the XML IPC request * \param[in] host The confirming peer attrd's uname */ void attrd_handle_confirmation(int callid, const char *host) { struct confirmation_action *action = NULL; GList *node = NULL; if (expected_confirmations == NULL) { return; } action = pcmk__intkey_table_lookup(expected_confirmations, callid); if (action == NULL) { return; } node = g_list_find_custom(action->respondents, host, (GCompareFunc) strcasecmp); if (node == NULL) { return; } action->respondents = g_list_remove(action->respondents, node->data); crm_trace("Callid %d now waiting on %d confirmations", callid, g_list_length(action->respondents)); if (action->respondents == NULL) { action->fn(action->xml); pcmk__intkey_table_remove(expected_confirmations, callid); crm_trace("%d requests now in expected confirmations table", g_hash_table_size(expected_confirmations)); } } diff --git a/daemons/attrd/pacemaker-attrd.h b/daemons/attrd/pacemaker-attrd.h index bcc329d8b1..83d7c6b0c5 100644 --- a/daemons/attrd/pacemaker-attrd.h +++ b/daemons/attrd/pacemaker-attrd.h @@ -1,208 +1,211 @@ /* * Copyright 2013-2022 the Pacemaker project contributors * * The version control history for this file may have further details. * * This source code is licensed under the GNU General Public License version 2 * or later (GPLv2+) WITHOUT ANY WARRANTY. */ #ifndef PACEMAKER_ATTRD__H # define PACEMAKER_ATTRD__H #include #include #include #include #include #include #include /* * Legacy attrd (all pre-1.1.11 Pacemaker versions, plus all versions when used * with the no-longer-supported CMAN or corosync-plugin stacks) is unversioned. * * With atomic attrd, each attrd will send ATTRD_PROTOCOL_VERSION with every * peer request and reply. As of Pacemaker 2.0.0, at start-up each attrd will * also set a private attribute for itself with its version, so any attrd can * determine the minimum version supported by all peers. * * Protocol Pacemaker Significant changes * -------- --------- ------------------- * 1 1.1.11 PCMK__ATTRD_CMD_UPDATE (PCMK__XA_ATTR_NAME only), * PCMK__ATTRD_CMD_PEER_REMOVE, PCMK__ATTRD_CMD_REFRESH, * PCMK__ATTRD_CMD_FLUSH, PCMK__ATTRD_CMD_SYNC, * PCMK__ATTRD_CMD_SYNC_RESPONSE * 1 1.1.13 PCMK__ATTRD_CMD_UPDATE (with PCMK__XA_ATTR_PATTERN), * PCMK__ATTRD_CMD_QUERY * 1 1.1.15 PCMK__ATTRD_CMD_UPDATE_BOTH, * PCMK__ATTRD_CMD_UPDATE_DELAY * 2 1.1.17 PCMK__ATTRD_CMD_CLEAR_FAILURE * 3 2.1.1 PCMK__ATTRD_CMD_SYNC_RESPONSE indicates remote nodes * 4 2.1.5 Multiple attributes can be updated in a single IPC * message * 5 2.1.5 Peers can request confirmation of a sent message */ #define ATTRD_PROTOCOL_VERSION "5" +#define ATTRD_SUPPORTS_MULTI_MESSAGE(x) ((x) >= 4) +#define ATTRD_SUPPORTS_CONFIRMATION(x) ((x) >= 5) + #define attrd_send_ack(client, id, flags) \ pcmk__ipc_send_ack((client), (id), (flags), "ack", ATTRD_PROTOCOL_VERSION, CRM_EX_INDETERMINATE) void attrd_init_mainloop(void); void attrd_run_mainloop(void); void attrd_set_requesting_shutdown(void); void attrd_clear_requesting_shutdown(void); void attrd_free_waitlist(void); bool attrd_requesting_shutdown(void); bool attrd_shutting_down(void); void attrd_shutdown(int nsig); void attrd_init_ipc(void); void attrd_ipc_fini(void); void attrd_cib_disconnect(void); bool attrd_value_needs_expansion(const char *value); int attrd_expand_value(const char *value, const char *old_value); /* regular expression to clear failures of all resources */ #define ATTRD_RE_CLEAR_ALL \ "^(" PCMK__FAIL_COUNT_PREFIX "|" PCMK__LAST_FAILURE_PREFIX ")-" /* regular expression to clear failure of all operations for one resource * (format takes resource name) * * @COMPAT attributes set < 1.1.17: * also match older attributes that do not have the operation part */ #define ATTRD_RE_CLEAR_ONE ATTRD_RE_CLEAR_ALL "%s(#.+_[0-9]+)?$" /* regular expression to clear failure of one operation for one resource * (format takes resource name, operation name, and interval) * * @COMPAT attributes set < 1.1.17: * also match older attributes that do not have the operation part */ #define ATTRD_RE_CLEAR_OP ATTRD_RE_CLEAR_ALL "%s(#%s_%u)?$" int attrd_failure_regex(regex_t *regex, const char *rsc, const char *op, guint interval_ms); extern cib_t *the_cib; /* Alerts */ extern lrmd_t *the_lrmd; extern crm_trigger_t *attrd_config_read; void attrd_lrmd_disconnect(void); gboolean attrd_read_options(gpointer user_data); void attrd_cib_replaced_cb(const char *event, xmlNode * msg); void attrd_cib_updated_cb(const char *event, xmlNode *msg); int attrd_send_attribute_alert(const char *node, int nodeid, const char *attr, const char *value); // Elections void attrd_election_init(void); void attrd_election_fini(void); void attrd_start_election_if_needed(void); bool attrd_election_won(void); void attrd_handle_election_op(const crm_node_t *peer, xmlNode *xml); bool attrd_check_for_new_writer(const crm_node_t *peer, const xmlNode *xml); void attrd_declare_winner(void); void attrd_remove_voter(const crm_node_t *peer); void attrd_xml_add_writer(xmlNode *xml); 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; gboolean force_write; /* Flag for updating attribute by ignoring delay */ } attribute_t; typedef struct attribute_value_s { uint32_t nodeid; gboolean is_remote; char *nodename; char *current; char *requested; gboolean seen; } attribute_value_t; extern crm_cluster_t *attrd_cluster; extern GHashTable *attributes; extern GHashTable *peer_protocol_vers; #define CIB_OP_TIMEOUT_S 120 int attrd_cluster_connect(void); void attrd_peer_update(const 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, bool uncache, const char *source); void attrd_peer_clear_failure(pcmk__request_t *request); void attrd_peer_sync_response(const crm_node_t *peer, bool peer_won, xmlNode *xml); void attrd_broadcast_protocol(void); xmlNode *attrd_client_peer_remove(pcmk__request_t *request); xmlNode *attrd_client_clear_failure(pcmk__request_t *request); xmlNode *attrd_client_update(pcmk__request_t *request); xmlNode *attrd_client_refresh(pcmk__request_t *request); xmlNode *attrd_client_query(pcmk__request_t *request); gboolean attrd_send_message(crm_node_t *node, xmlNode *data, bool confirm); xmlNode *attrd_add_value_xml(xmlNode *parent, const attribute_t *a, const attribute_value_t *v, bool force_write); void attrd_clear_value_seen(void); void attrd_free_attribute(gpointer data); void attrd_free_attribute_value(gpointer data); attribute_t *attrd_populate_attribute(xmlNode *xml, const char *attr); void attrd_write_attribute(attribute_t *a, bool ignore_delay); void attrd_write_attributes(bool all, bool ignore_delay); void attrd_write_or_elect_attribute(attribute_t *a); extern int minimum_protocol_version; void attrd_remove_peer_protocol_ver(const char *host); void attrd_update_minimum_protocol_ver(const char *host, const char *value); mainloop_timer_t *attrd_add_timer(const char *id, int timeout_ms, attribute_t *attr); void attrd_unregister_handlers(void); void attrd_handle_request(pcmk__request_t *request); enum attrd_sync_point { attrd_sync_point_local, attrd_sync_point_cluster, }; typedef int (*attrd_confirmation_action_fn)(xmlNode *); void attrd_add_client_to_waitlist(pcmk__request_t *request); void attrd_ack_waitlist_clients(enum attrd_sync_point sync_point, const xmlNode *xml); int attrd_cluster_sync_point_update(xmlNode *xml); void attrd_do_not_expect_from_peer(const char *host); void attrd_do_not_wait_for_client(pcmk__client_t *client); void attrd_expect_confirmations(pcmk__request_t *request, attrd_confirmation_action_fn fn); void attrd_free_confirmations(void); void attrd_handle_confirmation(int callid, const char *host); void attrd_remove_client_from_waitlist(pcmk__client_t *client); const char *attrd_request_sync_point(xmlNode *xml); bool attrd_request_has_sync_point(xmlNode *xml); #endif /* PACEMAKER_ATTRD__H */