16 #include <sys/types.h> 25 #define PCMK_IPC_DEFAULT_QUEUE_MAX 500 27 static GHashTable *client_connections = NULL;
38 return client_connections? g_hash_table_size(client_connections) : 0;
53 if ((func != NULL) && (client_connections != NULL)) {
54 g_hash_table_foreach(client_connections, func, user_data);
61 if (client_connections) {
62 return g_hash_table_lookup(client_connections, c);
72 if ((client_connections != NULL) && (
id != NULL)) {
77 g_hash_table_iter_init(&iter, client_connections);
78 while (g_hash_table_iter_next(&iter, &key, (gpointer *) & client)) {
79 if (strcmp(client->id,
id) == 0) {
84 crm_trace(
"No client found with id='%s'", pcmk__s(
id,
""));
101 return "(unspecified)";
103 }
else if (c->
name != NULL) {
106 }
else if (c->
id != NULL) {
110 return "(unidentified)";
117 if (client_connections != NULL) {
118 int active = g_hash_table_size(client_connections);
121 crm_warn(
"Exiting with %d active IPC client%s",
124 g_hash_table_destroy(client_connections);
125 client_connections = NULL;
132 qb_ipcs_connection_t *c = NULL;
134 if (service == NULL) {
138 c = qb_ipcs_connection_first_get(service);
141 qb_ipcs_connection_t *last = c;
143 c = qb_ipcs_connection_next_get(service, last);
146 crm_notice(
"Disconnecting client %p, pid=%d...",
148 qb_ipcs_disconnect(last);
149 qb_ipcs_connection_unref(last);
164 client_from_connection(qb_ipcs_connection_t *c,
void *key, uid_t uid_client)
170 if (client->
user == NULL) {
172 crm_err(
"Unable to enforce ACLs for user ID %d, assuming unprivileged",
187 if (client_connections == NULL) {
189 client_connections = g_hash_table_new(g_direct_hash, g_direct_equal);
191 g_hash_table_insert(client_connections, key, client);
205 return client_from_connection(NULL, key, 0);
211 gid_t uid_cluster = 0;
212 gid_t gid_cluster = 0;
219 static bool need_log = TRUE;
222 crm_warn(
"Could not find user and group IDs for user %s",
228 if (uid_client != 0) {
229 crm_trace(
"Giving group %u access to new IPC connection", gid_cluster);
231 qb_ipcs_connection_auth_set(c, -1, gid_cluster, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
235 client = client_from_connection(c, NULL, uid_client);
237 if ((uid_client == 0) || (uid_client == uid_cluster)) {
242 crm_debug(
"New IPC client %s for PID %u with uid %d and gid %d",
243 client->
id, client->
pid, uid_client, gid_client);
247 static struct iovec *
248 pcmk__new_ipc_event(
void)
262 free(event[0].iov_base);
263 free(event[1].iov_base);
269 free_event(gpointer
data)
290 if (client_connections) {
292 crm_trace(
"Destroying %p/%p (%d remaining)",
293 c, c->
ipcs, g_hash_table_size(client_connections) - 1);
294 g_hash_table_remove(client_connections, c->
ipcs);
297 crm_trace(
"Destroying remote connection %p (%d remaining)",
298 c, g_hash_table_size(client_connections) - 1);
299 g_hash_table_remove(client_connections, c->
id);
319 #ifdef HAVE_GNUTLS_GNUTLS_H 320 if (c->
remote->tls_session != NULL) {
324 gnutls_free(c->
remote->tls_session);
326 #endif // HAVE_GNUTLS_GNUTLS_H 344 long long qmax_ll = 0LL;
345 unsigned int orig_value = 0U;
354 if ((qmax_ll <= 0LL) || (qmax_ll > UINT_MAX)) {
357 client->
queue_max = (
unsigned int) qmax_ll;
365 crm_info(
"Could not set IPC threshold for client %s[%u] to %s: %s",
369 }
else if (client->
queue_max != orig_value) {
370 crm_debug(
"IPC threshold for client %s[%u] is now %u (was %u)",
379 struct qb_ipcs_connection_stats stats;
381 stats.client_pid = 0;
382 qb_ipcs_connection_stats_get(c, &stats, 0);
383 return stats.client_pid;
402 char *uncompressed = NULL;
411 *
id = ((
struct qb_ipc_response_header *)
data)->id;
430 crm_trace(
"Decompressing message data %u bytes into %u bytes",
433 rc = BZ2_bzBuffToBuffDecompress(uncompressed, &size_u, text, header->
size_compressed, 1, 0);
458 crm_ipcs_flush_events_cb(gpointer
data)
463 crm_ipcs_flush_events(c);
478 guint
delay = (queue_len < 5)? (1000 + 100 * queue_len) : 1500;
496 unsigned int sent = 0;
497 unsigned int queue_len = 0;
513 struct iovec *
event = NULL;
523 qb_rc = qb_ipcs_event_sendv(c->
ipcs, event, 2);
531 header =
event[0].iov_base;
533 crm_trace(
"Event %d to %p[%d] (%lld compressed bytes) sent",
534 header->
qb.id, c->
ipcs, c->
pid, (
long long) qb_rc);
536 crm_trace(
"Event %d to %p[%d] (%lld bytes) sent: %.120s",
537 header->
qb.id, c->
ipcs, c->
pid, (
long long) qb_rc,
538 (
char *) (event[1].iov_base));
544 if (sent > 0 || queue_len) {
545 crm_trace(
"Sent %d events (%d remaining) for %p[%d]: %s (%lld)",
546 sent, queue_len, c->
ipcs, c->
pid,
557 if ((c->
queue_backlog <= 1) || (queue_len < c->queue_backlog)) {
559 crm_warn(
"Client with process ID %u has a backlog of %u messages " 562 crm_err(
"Evicting client with process ID %u due to backlog of %u messages " 565 qb_ipcs_disconnect(c->
ipcs);
571 delay_next_flush(c, queue_len);
595 uint32_t max_send_size,
struct iovec **
result,
599 unsigned int total = 0;
600 GString *buffer = NULL;
604 if ((message == NULL) || (
result == NULL)) {
610 if (header == NULL) {
615 buffer = g_string_sized_new(1024);
618 if (max_send_size == 0) {
624 iov = pcmk__new_ipc_event();
626 iov[0].iov_base = header;
632 if (total < max_send_size) {
637 static unsigned int biggest = 0;
639 char *compressed = NULL;
640 unsigned int new_size = 0;
644 (
unsigned int) max_send_size, &compressed,
651 iov[1].iov_base = compressed;
659 crm_err(
"Could not compress %u-byte message into less than IPC " 660 "limit of %u bytes; set PCMK_ipc_buffer to higher value " 661 "(%u bytes suggested)",
671 header->
qb.size = iov[0].iov_len + iov[1].iov_len;
672 header->
qb.id = (int32_t)request;
677 *bytes = header->
qb.size;
681 if (buffer != NULL) {
682 g_string_free(buffer, TRUE);
691 static uint32_t
id = 1;
708 header->
qb.id =
id++;
715 struct iovec *iov_copy = pcmk__new_ipc_event();
718 iov_copy[0].iov_len = iov[0].iov_len;
719 iov_copy[0].iov_base = malloc(iov[0].iov_len);
720 memcpy(iov_copy[0].iov_base, iov[0].iov_base, iov[0].iov_len);
722 iov_copy[1].iov_len = iov[1].iov_len;
723 iov_copy[1].iov_base = malloc(iov[1].iov_len);
724 memcpy(iov_copy[1].iov_base, iov[1].iov_base, iov[1].iov_len);
726 add_event(c, iov_copy);
734 qb_rc = qb_ipcs_response_sendv(c->
ipcs, iov, 2);
735 if (qb_rc < header->qb.size) {
739 crm_notice(
"Response %d to pid %d failed: %s " 740 CRM_XS " bytes=%u rc=%lld ipcs=%p",
742 header->
qb.size, (
long long) qb_rc, c->
ipcs);
745 crm_trace(
"Response %d sent, %lld bytes to %p[%d]",
746 header->
qb.id, (
long long) qb_rc, c->
ipcs, c->
pid);
755 rc = crm_ipcs_flush_events(c);
757 crm_ipcs_flush_events(c);
760 if ((rc == EPIPE) || (rc == ENOTCONN)) {
770 struct iovec *iov = NULL;
805 const char *tag,
const char *ver,
crm_exit_t status)
836 uint32_t request, uint32_t
flags,
const char *tag,
843 crm_trace(
"Ack'ing IPC message from client %s as <%s status=%d>",
866 qb_ipcs_service_t **ipcs_rw,
867 qb_ipcs_service_t **ipcs_shm,
868 struct qb_ipcs_service_handlers *ro_cb,
869 struct qb_ipcs_service_handlers *rw_cb)
872 QB_IPC_NATIVE, ro_cb);
875 QB_IPC_NATIVE, rw_cb);
880 if (*ipcs_ro == NULL || *ipcs_rw == NULL || *ipcs_shm == NULL) {
881 crm_err(
"Failed to create the CIB manager: exiting and inhibiting respawn");
882 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled");
900 qb_ipcs_service_t *ipcs_rw,
901 qb_ipcs_service_t *ipcs_shm)
903 qb_ipcs_destroy(ipcs_ro);
904 qb_ipcs_destroy(ipcs_rw);
905 qb_ipcs_destroy(ipcs_shm);
933 struct qb_ipcs_service_handlers *cb)
938 crm_err(
"Failed to create pacemaker-attrd server: exiting and inhibiting respawn");
939 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled.");
955 struct qb_ipcs_service_handlers *cb)
961 crm_err(
"Failed to create fencer: exiting and inhibiting respawn.");
962 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled.");
978 struct qb_ipcs_service_handlers *cb)
983 crm_err(
"Couldn't start pacemakerd IPC server");
984 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled.");
1032 "pacemaker-remoted",
#define CRM_CHECK(expr, failure_action)
#define PCMK__SERVER_BASED_RW
void pcmk__drop_all_clients(qb_ipcs_service_t *service)
#define crm_notice(fmt, args...)
const char * pcmk__client_name(const pcmk__client_t *c)
char * crm_generate_uuid(void)
void pcmk__client_cleanup(void)
void pcmk__free_client(pcmk__client_t *c)
_Noreturn crm_exit_t crm_exit(crm_exit_t rc)
pcmk__client_t * pcmk__new_client(qb_ipcs_connection_t *c, uid_t uid_client, gid_t gid_client)
bool crm_is_daemon_name(const char *name)
Check whether string represents a client name used by cluster daemons.
#define PCMK__VALUE_ATTRD
char * pcmk__uid2username(uid_t uid)
void pcmk__xml_string(const xmlNode *data, uint32_t options, GString *buffer, int depth)
qb_ipcs_service_t * mainloop_add_ipc_server_with_prio(const char *name, enum qb_ipc_type type, struct qb_ipcs_service_handlers *callbacks, enum qb_loop_priority prio)
Start server-side API end-point, hooked into the internal event loop.
xmlNode * pcmk__client_data2xml(pcmk__client_t *c, void *data, uint32_t *id, uint32_t *flags)
int pcmk__client_pid(qb_ipcs_connection_t *c)
const char * crm_xml_add_int(xmlNode *node, const char *name, int value)
Create an XML attribute with specified name and integer value.
const char * crm_xml_add(xmlNode *node, const char *name, const char *value)
Create an XML attribute with specified name and value.
void pcmk_free_ipc_event(struct iovec *event)
Free an I/O vector created by pcmk__ipc_prepare_iov()
enum crm_exit_e crm_exit_t
#define CRM_LOG_ASSERT(expr)
int pcmk_daemon_user(uid_t *uid, gid_t *gid)
Get user and group IDs of pacemaker daemon user.
struct pcmk__ipc_header_s pcmk__ipc_header_t
const char * pcmk_rc_str(int rc)
Get a user-friendly description of a return code.
void pcmk__foreach_ipc_client(GHFunc func, gpointer user_data)
void pcmk__serve_attrd_ipc(qb_ipcs_service_t **ipcs, struct qb_ipcs_service_handlers *cb)
pcmk__client_t * pcmk__new_unauth_client(void *key)
Allocate a new pcmk__client_t object and generate its ID.
Client is run by root or cluster user.
#define pcmk__set_client_flags(client, flags_to_set)
void pcmk__serve_fenced_ipc(qb_ipcs_service_t **ipcs, struct qb_ipcs_service_handlers *cb)
int pcmk__scan_ll(const char *text, long long *result, long long default_value)
#define crm_warn(fmt, args...)
pcmk__client_t * pcmk__find_client_by_id(const char *id)
#define PCMK__IPC_VERSION
#define crm_debug(fmt, args...)
void pcmk__serve_pacemakerd_ipc(qb_ipcs_service_t **ipcs, struct qb_ipcs_service_handlers *cb)
void pcmk__set_client_queue_max(pcmk__client_t *client, const char *qmax)
External (OS/environmental) problem.
#define crm_trace(fmt, args...)
#define pcmk_is_set(g, f)
Convenience alias for pcmk_all_flags_set(), to check single flag.
#define CRM_SYSTEM_PENGINE
Wrappers for and extensions to libxml2.
G_GNUC_INTERNAL bool pcmk__valid_ipc_header(const pcmk__ipc_header_t *header)
int pcmk__compress(const char *data, unsigned int length, unsigned int max, char **result, unsigned int *result_len)
unsigned int crm_ipc_default_buffer_size(void)
Return pacemaker's default IPC buffer size.
pcmk__client_t * pcmk__find_client(const qb_ipcs_connection_t *c)
void free_xml(xmlNode *child)
#define PCMK__SERVER_BASED_RO
bool pcmk__str_any_of(const char *s,...) G_GNUC_NULL_TERMINATED
#define pcmk__str_copy(str)
xmlNode * pcmk__xml_parse(const char *input)
int pcmk__ipc_send_xml(pcmk__client_t *c, uint32_t request, const xmlNode *message, uint32_t flags)
const char * pcmk__message_name(const char *name)
Get name to be used as identifier for cluster messages.
#define PCMK__XA_IPC_PROTO_VERSION
#define pcmk__assert(expr)
qb_ipcs_service_t * pcmk__serve_controld_ipc(struct qb_ipcs_service_handlers *cb)
int pcmk__ipc_send_iov(pcmk__client_t *c, struct iovec *iov, uint32_t flags)
int pcmk__ipc_prepare_iov(uint32_t request, const xmlNode *message, uint32_t max_send_size, struct iovec **result, ssize_t *bytes)
#define CRM_SYSTEM_STONITHD
int pcmk__bzlib2rc(int bz2)
Map a bz2 return code to the most similar Pacemaker return code.
guint pcmk__ipc_client_count(void)
#define PCMK_IPC_DEFAULT_QUEUE_MAX
#define CRM_SYSTEM_TENGINE
pcmk__action_result_t result
xmlNode * pcmk__ipc_create_ack_as(const char *function, int line, uint32_t flags, const char *tag, const char *ver, crm_exit_t status)
#define pcmk__set_ipc_flags(ipc_flags, ipc_name, flags_to_set)
#define crm_err(fmt, args...)
void pcmk__serve_based_ipc(qb_ipcs_service_t **ipcs_ro, qb_ipcs_service_t **ipcs_rw, qb_ipcs_service_t **ipcs_shm, struct qb_ipcs_service_handlers *ro_cb, struct qb_ipcs_service_handlers *rw_cb)
int pcmk__ipc_send_ack_as(const char *function, int line, pcmk__client_t *c, uint32_t request, uint32_t flags, const char *tag, const char *ver, crm_exit_t status)
#define pcmk__plural_s(i)
IPC interface to Pacemaker daemons.
#define crm_log_xml_trace(xml, text)
#define PCMK__SERVER_BASED_SHM
struct pcmk__remote_s * remote
qb_ipcs_service_t * mainloop_add_ipc_server(const char *name, enum qb_ipc_type type, struct qb_ipcs_service_handlers *callbacks)
xmlNode * pcmk__xe_create(xmlNode *parent, const char *name)
#define pcmk__assert_alloc(nmemb, size)
void pcmk__stop_based_ipc(qb_ipcs_service_t *ipcs_ro, qb_ipcs_service_t *ipcs_rw, qb_ipcs_service_t *ipcs_shm)
unsigned int queue_backlog
#define crm_info(fmt, args...)
qb_ipcs_connection_t * ipcs
qb_ipcs_service_t * pcmk__serve_schedulerd_ipc(struct qb_ipcs_service_handlers *cb)