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 349 && (qmax_ll > 0LL) && (qmax_ll <= UINT_MAX)) {
350 client->
queue_max = (
unsigned int) qmax_ll;
360 struct qb_ipcs_connection_stats stats;
362 stats.client_pid = 0;
363 qb_ipcs_connection_stats_get(c, &stats, 0);
364 return stats.client_pid;
383 char *uncompressed = NULL;
392 *
id = ((
struct qb_ipc_response_header *)
data)->id;
411 crm_trace(
"Decompressing message data %u bytes into %u bytes",
414 rc = BZ2_bzBuffToBuffDecompress(uncompressed, &size_u, text, header->
size_compressed, 1, 0);
439 crm_ipcs_flush_events_cb(gpointer
data)
444 crm_ipcs_flush_events(c);
459 guint
delay = (queue_len < 5)? (1000 + 100 * queue_len) : 1500;
477 unsigned int sent = 0;
478 unsigned int queue_len = 0;
494 struct iovec *
event = NULL;
504 qb_rc = qb_ipcs_event_sendv(c->
ipcs, event, 2);
512 header =
event[0].iov_base;
514 crm_trace(
"Event %d to %p[%d] (%lld compressed bytes) sent",
515 header->
qb.id, c->
ipcs, c->
pid, (
long long) qb_rc);
517 crm_trace(
"Event %d to %p[%d] (%lld bytes) sent: %.120s",
518 header->
qb.id, c->
ipcs, c->
pid, (
long long) qb_rc,
519 (
char *) (event[1].iov_base));
525 if (sent > 0 || queue_len) {
526 crm_trace(
"Sent %d events (%d remaining) for %p[%d]: %s (%lld)",
527 sent, queue_len, c->
ipcs, c->
pid,
538 if ((c->
queue_backlog <= 1) || (queue_len < c->queue_backlog)) {
540 crm_warn(
"Client with process ID %u has a backlog of %u messages " 543 crm_err(
"Evicting client with process ID %u due to backlog of %u messages " 546 qb_ipcs_disconnect(c->
ipcs);
552 delay_next_flush(c, queue_len);
576 uint32_t max_send_size,
struct iovec **
result,
580 unsigned int total = 0;
581 GString *buffer = NULL;
585 if ((message == NULL) || (
result == NULL)) {
591 if (header == NULL) {
596 buffer = g_string_sized_new(1024);
599 if (max_send_size == 0) {
605 iov = pcmk__new_ipc_event();
607 iov[0].iov_base = header;
613 if (total < max_send_size) {
618 static unsigned int biggest = 0;
620 char *compressed = NULL;
621 unsigned int new_size = 0;
625 (
unsigned int) max_send_size, &compressed,
632 iov[1].iov_base = compressed;
640 crm_err(
"Could not compress %u-byte message into less than IPC " 641 "limit of %u bytes; set PCMK_ipc_buffer to higher value " 642 "(%u bytes suggested)",
652 header->
qb.size = iov[0].iov_len + iov[1].iov_len;
653 header->
qb.id = (int32_t)request;
658 *bytes = header->
qb.size;
662 if (buffer != NULL) {
663 g_string_free(buffer, TRUE);
672 static uint32_t
id = 1;
689 header->
qb.id =
id++;
696 struct iovec *iov_copy = pcmk__new_ipc_event();
699 iov_copy[0].iov_len = iov[0].iov_len;
700 iov_copy[0].iov_base = malloc(iov[0].iov_len);
701 memcpy(iov_copy[0].iov_base, iov[0].iov_base, iov[0].iov_len);
703 iov_copy[1].iov_len = iov[1].iov_len;
704 iov_copy[1].iov_base = malloc(iov[1].iov_len);
705 memcpy(iov_copy[1].iov_base, iov[1].iov_base, iov[1].iov_len);
707 add_event(c, iov_copy);
715 qb_rc = qb_ipcs_response_sendv(c->
ipcs, iov, 2);
716 if (qb_rc < header->qb.size) {
720 crm_notice(
"Response %d to pid %d failed: %s " 721 CRM_XS " bytes=%u rc=%lld ipcs=%p",
723 header->
qb.size, (
long long) qb_rc, c->
ipcs);
726 crm_trace(
"Response %d sent, %lld bytes to %p[%d]",
727 header->
qb.id, (
long long) qb_rc, c->
ipcs, c->
pid);
736 rc = crm_ipcs_flush_events(c);
738 crm_ipcs_flush_events(c);
741 if ((rc == EPIPE) || (rc == ENOTCONN)) {
751 struct iovec *iov = NULL;
786 const char *tag,
const char *ver,
crm_exit_t status)
817 uint32_t request, uint32_t
flags,
const char *tag,
824 crm_trace(
"Ack'ing IPC message from client %s as <%s status=%d>",
847 qb_ipcs_service_t **ipcs_rw,
848 qb_ipcs_service_t **ipcs_shm,
849 struct qb_ipcs_service_handlers *ro_cb,
850 struct qb_ipcs_service_handlers *rw_cb)
853 QB_IPC_NATIVE, ro_cb);
856 QB_IPC_NATIVE, rw_cb);
861 if (*ipcs_ro == NULL || *ipcs_rw == NULL || *ipcs_shm == NULL) {
862 crm_err(
"Failed to create the CIB manager: exiting and inhibiting respawn");
863 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled");
881 qb_ipcs_service_t *ipcs_rw,
882 qb_ipcs_service_t *ipcs_shm)
884 qb_ipcs_destroy(ipcs_ro);
885 qb_ipcs_destroy(ipcs_rw);
886 qb_ipcs_destroy(ipcs_shm);
914 struct qb_ipcs_service_handlers *cb)
919 crm_err(
"Failed to create pacemaker-attrd server: exiting and inhibiting respawn");
920 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled.");
936 struct qb_ipcs_service_handlers *cb)
942 crm_err(
"Failed to create fencer: exiting and inhibiting respawn.");
943 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled.");
959 struct qb_ipcs_service_handlers *cb)
964 crm_err(
"Couldn't start pacemakerd IPC server");
965 crm_warn(
"Verify pacemaker and pacemaker_remote are not both enabled.");
1013 "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)
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
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
bool pcmk__set_client_queue_max(pcmk__client_t *client, const char *qmax)
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
qb_ipcs_connection_t * ipcs
qb_ipcs_service_t * pcmk__serve_schedulerd_ipc(struct qb_ipcs_service_handlers *cb)