lib/common/ipc

/* */
This source file includes following definitions.
pcmk__ipc_client_count
pcmk__foreach_ipc_client
pcmk__find_client
pcmk__find_client_by_id
pcmk__client_name
pcmk__client_cleanup
pcmk__drop_all_clients
client_from_connection
pcmk__new_unauth_client
pcmk__new_client
pcmk__new_ipc_event
pcmk_free_ipc_event
free_event
add_event
pcmk__free_client
pcmk__set_client_queue_max
pcmk__client_pid
pcmk__client_data2xml
crm_ipcs_flush_events_cb
delay_next_flush
crm_ipcs_flush_events
pcmk__ipc_prepare_iov
pcmk__ipc_send_iov
pcmk__ipc_send_xml
pcmk__ipc_send_ack_as
pcmk__serve_based_ipc
pcmk__stop_based_ipc
pcmk__serve_controld_ipc
pcmk__serve_attrd_ipc
pcmk__serve_fenced_ipc
crm_is_daemon_name
   1 /*
   2  * Copyright 2004-2020 the Pacemaker project contributors
   3  *
   4  * The version control history for this file may have further details.
   5  *
   6  * This source code is licensed under the GNU Lesser General Public License
   7  * version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
   8  */
   9 
  10 #include <crm_internal.h>
  11 
  12 #include <stdio.h>
  13 #include <errno.h>
  14 #include <bzlib.h>
  15 #include <sys/stat.h>
  16 #include <sys/types.h>
  17 
  18 #include <crm/crm.h>
  19 #include <crm/msg_xml.h>
  20 #include <crm/common/ipc.h>
  21 #include <crm/common/ipc_internal.h>
  22 #include "crmcommon_private.h"
  23 
  24 /* Evict clients whose event queue grows this large (by default) */
  25 #define PCMK_IPC_DEFAULT_QUEUE_MAX 500
  26 
  27 static GHashTable *client_connections = NULL;
  28 
  29 /*!
  30  * \internal
  31  * \brief Count IPC clients
  32  *
  33  * \return Number of active IPC client connections
  34  */
  35 guint
  36 pcmk__ipc_client_count()
     /*  */
  37 {
  38     return client_connections? g_hash_table_size(client_connections) : 0;
  39 }
  40 
  41 /*!
  42  * \internal
  43  * \brief Execute a function for each active IPC client connection
  44  *
  45  * \param[in] func       Function to call
  46  * \param[in] user_data  Pointer to pass to function
  47  *
  48  * \note The parameters are the same as for g_hash_table_foreach().
  49  */
  50 void
  51 pcmk__foreach_ipc_client(GHFunc func, gpointer user_data)
     /*  */
  52 {
  53     if ((func != NULL) && (client_connections != NULL)) {
  54         g_hash_table_foreach(client_connections, func, user_data);
  55     }
  56 }
  57 
  58 pcmk__client_t *
  59 pcmk__find_client(qb_ipcs_connection_t *c)
     /*  */
  60 {
  61     if (client_connections) {
  62         return g_hash_table_lookup(client_connections, c);
  63     }
  64 
  65     crm_trace("No client found for %p", c);
  66     return NULL;
  67 }
  68 
  69 pcmk__client_t *
  70 pcmk__find_client_by_id(const char *id)
     /*  */
  71 {
  72     gpointer key;
  73     pcmk__client_t *client;
  74     GHashTableIter iter;
  75 
  76     if (client_connections && id) {
  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) {
  80                 return client;
  81             }
  82         }
  83     }
  84 
  85     crm_trace("No client found with id=%s", id);
  86     return NULL;
  87 }
  88 
  89 const char *
  90 pcmk__client_name(pcmk__client_t *c)
     /*  */
  91 {
  92     if (c == NULL) {
  93         return "null";
  94     } else if (c->name == NULL && c->id == NULL) {
  95         return "unknown";
  96     } else if (c->name == NULL) {
  97         return c->id;
  98     } else {
  99         return c->name;
 100     }
 101 }
 102 
 103 void
 104 pcmk__client_cleanup(void)
     /*  */
 105 {
 106     if (client_connections != NULL) {
 107         int active = g_hash_table_size(client_connections);
 108 
 109         if (active) {
 110             crm_err("Exiting with %d active IPC client%s",
 111                     active, pcmk__plural_s(active));
 112         }
 113         g_hash_table_destroy(client_connections); client_connections = NULL;
 114     }
 115 }
 116 
 117 void
 118 pcmk__drop_all_clients(qb_ipcs_service_t *service)
     /*  */
 119 {
 120     qb_ipcs_connection_t *c = NULL;
 121 
 122     if (service == NULL) {
 123         return;
 124     }
 125 
 126     c = qb_ipcs_connection_first_get(service);
 127 
 128     while (c != NULL) {
 129         qb_ipcs_connection_t *last = c;
 130 
 131         c = qb_ipcs_connection_next_get(service, last);
 132 
 133         /* There really shouldn't be anyone connected at this point */
 134         crm_notice("Disconnecting client %p, pid=%d...",
 135                    last, pcmk__client_pid(last));
 136         qb_ipcs_disconnect(last);
 137         qb_ipcs_connection_unref(last);
 138     }
 139 }
 140 
 141 /*!
 142  * \internal
 143  * \brief Allocate a new pcmk__client_t object based on an IPC connection
 144  *
 145  * \param[in] c           IPC connection (or NULL to allocate generic client)
 146  * \param[in] key         Connection table key (or NULL to use sane default)
 147  * \param[in] uid_client  UID corresponding to c (ignored if c is NULL)
 148  *
 149  * \return Pointer to new pcmk__client_t (or NULL on error)
 150  */
 151 static pcmk__client_t *
 152 client_from_connection(qb_ipcs_connection_t *c, void *key, uid_t uid_client)
     /*  */
 153 {
 154     pcmk__client_t *client = calloc(1, sizeof(pcmk__client_t));
 155 
 156     if (client == NULL) {
 157         crm_perror(LOG_ERR, "Allocating client");
 158         return NULL;
 159     }
 160 
 161     if (c) {
 162 #if ENABLE_ACL
 163         client->user = pcmk__uid2username(uid_client);
 164         if (client->user == NULL) {
 165             client->user = strdup("#unprivileged");
 166             CRM_CHECK(client->user != NULL, free(client); return NULL);
 167             crm_err("Unable to enforce ACLs for user ID %d, assuming unprivileged",
 168                     uid_client);
 169         }
 170 #endif
 171         client->ipcs = c;
 172         pcmk__set_client_flags(client, pcmk__client_ipc);
 173         client->pid = pcmk__client_pid(c);
 174         if (key == NULL) {
 175             key = c;
 176         }
 177     }
 178 
 179     client->id = crm_generate_uuid();
 180     if (client->id == NULL) {
 181         crm_err("Could not generate UUID for client");
 182         free(client->user);
 183         free(client);
 184         return NULL;
 185     }
 186     if (key == NULL) {
 187         key = client->id;
 188     }
 189     if (client_connections == NULL) {
 190         crm_trace("Creating IPC client table");
 191         client_connections = g_hash_table_new(g_direct_hash, g_direct_equal);
 192     }
 193     g_hash_table_insert(client_connections, key, client);
 194     return client;
 195 }
 196 
 197 /*!
 198  * \brief Allocate a new pcmk__client_t object and generate its ID
 199  *
 200  * \param[in] key  What to use as connections hash table key (NULL to use ID)
 201  *
 202  * \return Pointer to new pcmk__client_t (asserts on failure)
 203  */
 204 pcmk__client_t *
 205 pcmk__new_unauth_client(void *key)
     /*  */
 206 {
 207     pcmk__client_t *client = client_from_connection(NULL, key, 0);
 208 
 209     CRM_ASSERT(client != NULL);
 210     return client;
 211 }
 212 
 213 pcmk__client_t *
 214 pcmk__new_client(qb_ipcs_connection_t *c, uid_t uid_client, gid_t gid_client)
     /*  */
 215 {
 216     gid_t uid_cluster = 0;
 217     gid_t gid_cluster = 0;
 218 
 219     pcmk__client_t *client = NULL;
 220 
 221     CRM_CHECK(c != NULL, return NULL);
 222 
 223     if (pcmk_daemon_user(&uid_cluster, &gid_cluster) < 0) {
 224         static bool need_log = TRUE;
 225 
 226         if (need_log) {
 227             crm_warn("Could not find user and group IDs for user %s",
 228                      CRM_DAEMON_USER);
 229             need_log = FALSE;
 230         }
 231     }
 232 
 233     if (uid_client != 0) {
 234         crm_trace("Giving group %u access to new IPC connection", gid_cluster);
 235         /* Passing -1 to chown(2) means don't change */
 236         qb_ipcs_connection_auth_set(c, -1, gid_cluster, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
 237     }
 238 
 239     /* TODO: Do our own auth checking, return NULL if unauthorized */
 240     client = client_from_connection(c, NULL, uid_client);
 241     if (client == NULL) {
 242         return NULL;
 243     }
 244 
 245     if ((uid_client == 0) || (uid_client == uid_cluster)) {
 246         /* Remember when a connection came from root or hacluster */
 247         pcmk__set_client_flags(client, pcmk__client_privileged);
 248     }
 249 
 250     crm_debug("New IPC client %s for PID %u with uid %d and gid %d",
 251               client->id, client->pid, uid_client, gid_client);
 252     return client;
 253 }
 254 
 255 static struct iovec *
 256 pcmk__new_ipc_event(void)
     /*  */
 257 {
 258     struct iovec *iov = calloc(2, sizeof(struct iovec));
 259 
 260     CRM_ASSERT(iov != NULL);
 261     return iov;
 262 }
 263 
 264 /*!
 265  * \brief Free an I/O vector created by pcmk__ipc_prepare_iov()
 266  *
 267  * \param[in] event  I/O vector to free
 268  */
 269 void
 270 pcmk_free_ipc_event(struct iovec *event)
     /*  */
 271 {
 272     if (event != NULL) {
 273         free(event[0].iov_base);
 274         free(event[1].iov_base);
 275         free(event);
 276     }
 277 }
 278 
 279 static void
 280 free_event(gpointer data)
     /*  */
 281 {
 282     pcmk_free_ipc_event((struct iovec *) data);
 283 }
 284 
 285 static void
 286 add_event(pcmk__client_t *c, struct iovec *iov)
     /*  */
 287 {
 288     if (c->event_queue == NULL) {
 289         c->event_queue = g_queue_new();
 290     }
 291     g_queue_push_tail(c->event_queue, iov);
 292 }
 293 
 294 void
 295 pcmk__free_client(pcmk__client_t *c)
     /*  */
 296 {
 297     if (c == NULL) {
 298         return;
 299     }
 300 
 301     if (client_connections) {
 302         if (c->ipcs) {
 303             crm_trace("Destroying %p/%p (%d remaining)",
 304                       c, c->ipcs, g_hash_table_size(client_connections) - 1);
 305             g_hash_table_remove(client_connections, c->ipcs);
 306 
 307         } else {
 308             crm_trace("Destroying remote connection %p (%d remaining)",
 309                       c, g_hash_table_size(client_connections) - 1);
 310             g_hash_table_remove(client_connections, c->id);
 311         }
 312     }
 313 
 314     if (c->event_timer) {
 315         g_source_remove(c->event_timer);
 316     }
 317 
 318     if (c->event_queue) {
 319         crm_debug("Destroying %d events", g_queue_get_length(c->event_queue));
 320         g_queue_free_full(c->event_queue, free_event);
 321     }
 322 
 323     free(c->id);
 324     free(c->name);
 325     free(c->user);
 326     if (c->remote) {
 327         if (c->remote->auth_timeout) {
 328             g_source_remove(c->remote->auth_timeout);
 329         }
 330         free(c->remote->buffer);
 331         free(c->remote);
 332     }
 333     free(c);
 334 }
 335 
 336 /*!
 337  * \internal
 338  * \brief Raise IPC eviction threshold for a client, if allowed
 339  *
 340  * \param[in,out] client     Client to modify
 341  * \param[in]     qmax       New threshold (as non-NULL string)
 342  *
 343  * \return TRUE if change was allowed, FALSE otherwise
 344  */
 345 bool
 346 pcmk__set_client_queue_max(pcmk__client_t *client, const char *qmax)
     /*  */
 347 {
 348     if (pcmk_is_set(client->flags, pcmk__client_privileged)) {
 349         long long qmax_int;
 350 
 351         errno = 0;
 352         qmax_int = crm_parse_ll(qmax, NULL);
 353         if ((errno == 0) && (qmax_int > 0)) {
 354             client->queue_max = (unsigned int) qmax_int;
 355             return TRUE;
 356         }
 357     }
 358     return FALSE;
 359 }
 360 
 361 int
 362 pcmk__client_pid(qb_ipcs_connection_t *c)
     /*  */
 363 {
 364     struct qb_ipcs_connection_stats stats;
 365 
 366     stats.client_pid = 0;
 367     qb_ipcs_connection_stats_get(c, &stats, 0);
 368     return stats.client_pid;
 369 }
 370 
 371 /*!
 372  * \internal
 373  * \brief Retrieve message XML from data read from client IPC
 374  *
 375  * \param[in]  c       IPC client connection
 376  * \param[in]  data    Data read from client connection
 377  * \param[out] id      Where to store message ID from libqb header
 378  * \param[out] flags   Where to store flags from libqb header
 379  *
 380  * \return Message XML on success, NULL otherwise
 381  */
 382 xmlNode *
 383 pcmk__client_data2xml(pcmk__client_t *c, void *data, uint32_t *id,
     /*  */
 384                       uint32_t *flags)
 385 {
 386     xmlNode *xml = NULL;
 387     char *uncompressed = NULL;
 388     char *text = ((char *)data) + sizeof(pcmk__ipc_header_t);
 389     pcmk__ipc_header_t *header = data;
 390 
 391     if (!pcmk__valid_ipc_header(header)) {
 392         return NULL;
 393     }
 394 
 395     if (id) {
 396         *id = ((struct qb_ipc_response_header *)data)->id;
 397     }
 398     if (flags) {
 399         *flags = header->flags;
 400     }
 401 
 402     if (pcmk_is_set(header->flags, crm_ipc_proxied)) {
 403         /* Mark this client as being the endpoint of a proxy connection.
 404          * Proxy connections responses are sent on the event channel, to avoid
 405          * blocking the controller serving as proxy.
 406          */
 407         pcmk__set_client_flags(c, pcmk__client_proxied);
 408     }
 409 
 410     if (header->size_compressed) {
 411         int rc = 0;
 412         unsigned int size_u = 1 + header->size_uncompressed;
 413         uncompressed = calloc(1, size_u);
 414 
 415         crm_trace("Decompressing message data %u bytes into %u bytes",
 416                   header->size_compressed, size_u);
 417 
 418         rc = BZ2_bzBuffToBuffDecompress(uncompressed, &size_u, text, header->size_compressed, 1, 0);
 419         text = uncompressed;
 420 
 421         if (rc != BZ_OK) {
 422             crm_err("Decompression failed: %s " CRM_XS " bzerror=%d",
 423                     bz2_strerror(rc), rc);
 424             free(uncompressed);
 425             return NULL;
 426         }
 427     }
 428 
 429     CRM_ASSERT(text[header->size_uncompressed - 1] == 0);
 430 
 431     xml = string2xml(text);
 432     crm_log_xml_trace(xml, "[IPC received]");
 433 
 434     free(uncompressed);
 435     return xml;
 436 }
 437 
 438 static int crm_ipcs_flush_events(pcmk__client_t *c);
 439 
 440 static gboolean
 441 crm_ipcs_flush_events_cb(gpointer data)
     /*  */
 442 {
 443     pcmk__client_t *c = data;
 444 
 445     c->event_timer = 0;
 446     crm_ipcs_flush_events(c);
 447     return FALSE;
 448 }
 449 
 450 /*!
 451  * \internal
 452  * \brief Add progressive delay before next event queue flush
 453  *
 454  * \param[in,out] c          Client connection to add delay to
 455  * \param[in]     queue_len  Current event queue length
 456  */
 457 static inline void
 458 delay_next_flush(pcmk__client_t *c, unsigned int queue_len)
     /*  */
 459 {
 460     /* Delay a maximum of 1.5 seconds */
 461     guint delay = (queue_len < 5)? (1000 + 100 * queue_len) : 1500;
 462 
 463     c->event_timer = g_timeout_add(delay, crm_ipcs_flush_events_cb, c);
 464 }
 465 
 466 /*!
 467  * \internal
 468  * \brief Send client any messages in its queue
 469  *
 470  * \param[in]  c  Client to flush
 471  *
 472  * \return Standard Pacemaker return value
 473  */
 474 static int
 475 crm_ipcs_flush_events(pcmk__client_t *c)
     /*  */
 476 {
 477     int rc = pcmk_rc_ok;
 478     ssize_t qb_rc = 0;
 479     unsigned int sent = 0;
 480     unsigned int queue_len = 0;
 481 
 482     if (c == NULL) {
 483         return rc;
 484 
 485     } else if (c->event_timer) {
 486         /* There is already a timer, wait until it goes off */
 487         crm_trace("Timer active for %p - %d", c->ipcs, c->event_timer);
 488         return rc;
 489     }
 490 
 491     if (c->event_queue) {
 492         queue_len = g_queue_get_length(c->event_queue);
 493     }
 494     while (sent < 100) {
 495         pcmk__ipc_header_t *header = NULL;
 496         struct iovec *event = NULL;
 497 
 498         if (c->event_queue) {
 499             // We don't pop unless send is successful
 500             event = g_queue_peek_head(c->event_queue);
 501         }
 502         if (event == NULL) { // Queue is empty
 503             break;
 504         }
 505 
 506         qb_rc = qb_ipcs_event_sendv(c->ipcs, event, 2);
 507         if (qb_rc < 0) {
 508             rc = (int) -qb_rc;
 509             break;
 510         }
 511         event = g_queue_pop_head(c->event_queue);
 512 
 513         sent++;
 514         header = event[0].iov_base;
 515         if (header->size_compressed) {
 516             crm_trace("Event %d to %p[%d] (%lld compressed bytes) sent",
 517                       header->qb.id, c->ipcs, c->pid, (long long) qb_rc);
 518         } else {
 519             crm_trace("Event %d to %p[%d] (%lld bytes) sent: %.120s",
 520                       header->qb.id, c->ipcs, c->pid, (long long) qb_rc,
 521                       (char *) (event[1].iov_base));
 522         }
 523         pcmk_free_ipc_event(event);
 524     }
 525 
 526     queue_len -= sent;
 527     if (sent > 0 || queue_len) {
 528         crm_trace("Sent %d events (%d remaining) for %p[%d]: %s (%lld)",
 529                   sent, queue_len, c->ipcs, c->pid,
 530                   pcmk_rc_str(rc), (long long) qb_rc);
 531     }
 532 
 533     if (queue_len) {
 534 
 535         /* Allow clients to briefly fall behind on processing incoming messages,
 536          * but drop completely unresponsive clients so the connection doesn't
 537          * consume resources indefinitely.
 538          */
 539         if (queue_len > QB_MAX(c->queue_max, PCMK_IPC_DEFAULT_QUEUE_MAX)) {
 540             if ((c->queue_backlog <= 1) || (queue_len < c->queue_backlog)) {
 541                 /* Don't evict for a new or shrinking backlog */
 542                 crm_warn("Client with process ID %u has a backlog of %u messages "
 543                          CRM_XS " %p", c->pid, queue_len, c->ipcs);
 544             } else {
 545                 crm_err("Evicting client with process ID %u due to backlog of %u messages "
 546                          CRM_XS " %p", c->pid, queue_len, c->ipcs);
 547                 c->queue_backlog = 0;
 548                 qb_ipcs_disconnect(c->ipcs);
 549                 return rc;
 550             }
 551         }
 552 
 553         c->queue_backlog = queue_len;
 554         delay_next_flush(c, queue_len);
 555 
 556     } else {
 557         /* Event queue is empty, there is no backlog */
 558         c->queue_backlog = 0;
 559     }
 560 
 561     return rc;
 562 }
 563 
 564 /*!
 565  * \internal
 566  * \brief Create an I/O vector for sending an IPC XML message
 567  *
 568  * \param[in]  request        Identifier for libqb response header
 569  * \param[in]  message        XML message to send
 570  * \param[in]  max_send_size  If 0, default IPC buffer size is used
 571  * \param[out] result         Where to store prepared I/O vector
 572  * \param[out] bytes          Size of prepared data in bytes
 573  *
 574  * \return Standard Pacemaker return code
 575  */
 576 int
 577 pcmk__ipc_prepare_iov(uint32_t request, xmlNode *message,
     /*  */
 578                       uint32_t max_send_size, struct iovec **result,
 579                       ssize_t *bytes)
 580 {
 581     static unsigned int biggest = 0;
 582     struct iovec *iov;
 583     unsigned int total = 0;
 584     char *compressed = NULL;
 585     char *buffer = NULL;
 586     pcmk__ipc_header_t *header = NULL;
 587 
 588     if ((message == NULL) || (result == NULL)) {
 589         return EINVAL;
 590     }
 591 
 592     header = calloc(1, sizeof(pcmk__ipc_header_t));
 593     if (header == NULL) {
 594        return ENOMEM; /* errno mightn't be set by allocator */
 595     }
 596 
 597     buffer = dump_xml_unformatted(message);
 598 
 599     if (max_send_size == 0) {
 600         max_send_size = crm_ipc_default_buffer_size();
 601     }
 602     CRM_LOG_ASSERT(max_send_size != 0);
 603 
 604     *result = NULL;
 605     iov = pcmk__new_ipc_event();
 606     iov[0].iov_len = sizeof(pcmk__ipc_header_t);
 607     iov[0].iov_base = header;
 608 
 609     header->version = PCMK__IPC_VERSION;
 610     header->size_uncompressed = 1 + strlen(buffer);
 611     total = iov[0].iov_len + header->size_uncompressed;
 612 
 613     if (total < max_send_size) {
 614         iov[1].iov_base = buffer;
 615         iov[1].iov_len = header->size_uncompressed;
 616 
 617     } else {
 618         unsigned int new_size = 0;
 619 
 620         if (pcmk__compress(buffer, (unsigned int) header->size_uncompressed,
 621                            (unsigned int) max_send_size, &compressed,
 622                            &new_size) == pcmk_rc_ok) {
 623 
 624             pcmk__set_ipc_flags(header->flags, "send data", crm_ipc_compressed);
 625             header->size_compressed = new_size;
 626 
 627             iov[1].iov_len = header->size_compressed;
 628             iov[1].iov_base = compressed;
 629 
 630             free(buffer);
 631 
 632             biggest = QB_MAX(header->size_compressed, biggest);
 633 
 634         } else {
 635             crm_log_xml_trace(message, "EMSGSIZE");
 636             biggest = QB_MAX(header->size_uncompressed, biggest);
 637 
 638             crm_err("Could not compress %u-byte message into less than IPC "
 639                     "limit of %u bytes; set PCMK_ipc_buffer to higher value "
 640                     "(%u bytes suggested)",
 641                     header->size_uncompressed, max_send_size, 4 * biggest);
 642 
 643             free(compressed);
 644             free(buffer);
 645             pcmk_free_ipc_event(iov);
 646             return EMSGSIZE;
 647         }
 648     }
 649 
 650     header->qb.size = iov[0].iov_len + iov[1].iov_len;
 651     header->qb.id = (int32_t)request;    /* Replying to a specific request */
 652 
 653     *result = iov;
 654     CRM_ASSERT(header->qb.size > 0);
 655     if (bytes != NULL) {
 656         *bytes = header->qb.size;
 657     }
 658     return pcmk_rc_ok;
 659 }
 660 
 661 int
 662 pcmk__ipc_send_iov(pcmk__client_t *c, struct iovec *iov, uint32_t flags)
     /*  */
 663 {
 664     int rc = pcmk_rc_ok;
 665     static uint32_t id = 1;
 666     pcmk__ipc_header_t *header = iov[0].iov_base;
 667 
 668     if (c->flags & pcmk__client_proxied) {
 669         /* _ALL_ replies to proxied connections need to be sent as events */
 670         if (!pcmk_is_set(flags, crm_ipc_server_event)) {
 671             /* The proxied flag lets us know this was originally meant to be a
 672              * response, even though we're sending it over the event channel.
 673              */
 674             pcmk__set_ipc_flags(flags, "server event",
 675                                 crm_ipc_server_event
 676                                 |crm_ipc_proxied_relay_response);
 677         }
 678     }
 679 
 680     pcmk__set_ipc_flags(header->flags, "server event", flags);
 681     if (flags & crm_ipc_server_event) {
 682         header->qb.id = id++;   /* We don't really use it, but doesn't hurt to set one */
 683 
 684         if (flags & crm_ipc_server_free) {
 685             crm_trace("Sending the original to %p[%d]", c->ipcs, c->pid);
 686             add_event(c, iov);
 687 
 688         } else {
 689             struct iovec *iov_copy = pcmk__new_ipc_event();
 690 
 691             crm_trace("Sending a copy to %p[%d]", c->ipcs, c->pid);
 692             iov_copy[0].iov_len = iov[0].iov_len;
 693             iov_copy[0].iov_base = malloc(iov[0].iov_len);
 694             memcpy(iov_copy[0].iov_base, iov[0].iov_base, iov[0].iov_len);
 695 
 696             iov_copy[1].iov_len = iov[1].iov_len;
 697             iov_copy[1].iov_base = malloc(iov[1].iov_len);
 698             memcpy(iov_copy[1].iov_base, iov[1].iov_base, iov[1].iov_len);
 699 
 700             add_event(c, iov_copy);
 701         }
 702 
 703     } else {
 704         ssize_t qb_rc;
 705 
 706         CRM_LOG_ASSERT(header->qb.id != 0);     /* Replying to a specific request */
 707 
 708         qb_rc = qb_ipcs_response_sendv(c->ipcs, iov, 2);
 709         if (qb_rc < header->qb.size) {
 710             if (qb_rc < 0) {
 711                 rc = (int) -qb_rc;
 712             }
 713             crm_notice("Response %d to pid %d failed: %s "
 714                        CRM_XS " bytes=%u rc=%lld ipcs=%p",
 715                        header->qb.id, c->pid, pcmk_rc_str(rc),
 716                        header->qb.size, (long long) qb_rc, c->ipcs);
 717 
 718         } else {
 719             crm_trace("Response %d sent, %lld bytes to %p[%d]",
 720                       header->qb.id, (long long) qb_rc, c->ipcs, c->pid);
 721         }
 722 
 723         if (flags & crm_ipc_server_free) {
 724             pcmk_free_ipc_event(iov);
 725         }
 726     }
 727 
 728     if (flags & crm_ipc_server_event) {
 729         rc = crm_ipcs_flush_events(c);
 730     } else {
 731         crm_ipcs_flush_events(c);
 732     }
 733 
 734     if ((rc == EPIPE) || (rc == ENOTCONN)) {
 735         crm_trace("Client %p disconnected", c->ipcs);
 736     }
 737     return rc;
 738 }
 739 
 740 int
 741 pcmk__ipc_send_xml(pcmk__client_t *c, uint32_t request, xmlNode *message,
     /*  */
 742                    uint32_t flags)
 743 {
 744     struct iovec *iov = NULL;
 745     int rc = pcmk_rc_ok;
 746 
 747     if (c == NULL) {
 748         return EINVAL;
 749     }
 750     rc = pcmk__ipc_prepare_iov(request, message, crm_ipc_default_buffer_size(),
 751                                &iov, NULL);
 752     if (rc == pcmk_rc_ok) {
 753         pcmk__set_ipc_flags(flags, "send data", crm_ipc_server_free);
 754         rc = pcmk__ipc_send_iov(c, iov, flags);
 755     } else {
 756         pcmk_free_ipc_event(iov);
 757         crm_notice("IPC message to pid %d failed: %s " CRM_XS " rc=%d",
 758                    c->pid, pcmk_rc_str(rc), rc);
 759     }
 760     return rc;
 761 }
 762 
 763 /*!
 764  * \internal
 765  * \brief Send an acknowledgement with a status code to a client
 766  *
 767  * \param[in] function  Calling function
 768  * \param[in] line      Source file line within calling function
 769  * \param[in] c         Client to send ack to
 770  * \param[in] request   Request ID being replied to
 771  * \param[in] status    Exit status code to add to ack
 772  * \param[in] flags     IPC flags to use when sending
 773  * \param[in] tag       Element name to use for acknowledgement
 774  * \param[in] status    Status code to send with acknowledgement
 775  *
 776  * \return Standard Pacemaker return code
 777  */
 778 int
 779 pcmk__ipc_send_ack_as(const char *function, int line, pcmk__client_t *c,
     /*  */
 780                       uint32_t request, uint32_t flags, const char *tag,
 781                       crm_exit_t status)
 782 {
 783     int rc = pcmk_rc_ok;
 784 
 785     if (pcmk_is_set(flags, crm_ipc_client_response)) {
 786         xmlNode *ack = create_xml_node(NULL, tag);
 787 
 788         crm_trace("Ack'ing IPC message from %s as <%s status=%d>",
 789                   pcmk__client_name(c), tag, status);
 790         c->request_id = 0;
 791         crm_xml_add(ack, "function", function);
 792         crm_xml_add_int(ack, "line", line);
 793         crm_xml_add_int(ack, "status", (int) status);
 794         rc = pcmk__ipc_send_xml(c, request, ack, flags);
 795         free_xml(ack);
 796     }
 797     return rc;
 798 }
 799 
 800 /*!
 801  * \internal
 802  * \brief Add an IPC server to the main loop for the pacemaker-based API
 803  *
 804  * \param[out] ipcs_ro   New IPC server for read-only pacemaker-based API
 805  * \param[out] ipcs_rw   New IPC server for read/write pacemaker-based API
 806  * \param[out] ipcs_shm  New IPC server for shared-memory pacemaker-based API
 807  * \param[in]  ro_cb     IPC callbacks for read-only API
 808  * \param[in]  rw_cb     IPC callbacks for read/write and shared-memory APIs
 809  *
 810  * \note This function exits fatally if unable to create the servers.
 811  */
 812 void pcmk__serve_based_ipc(qb_ipcs_service_t **ipcs_ro,
     /*  */
 813                            qb_ipcs_service_t **ipcs_rw,
 814                            qb_ipcs_service_t **ipcs_shm,
 815                            struct qb_ipcs_service_handlers *ro_cb,
 816                            struct qb_ipcs_service_handlers *rw_cb)
 817 {
 818     *ipcs_ro = mainloop_add_ipc_server(PCMK__SERVER_BASED_RO,
 819                                        QB_IPC_NATIVE, ro_cb);
 820 
 821     *ipcs_rw = mainloop_add_ipc_server(PCMK__SERVER_BASED_RW,
 822                                        QB_IPC_NATIVE, rw_cb);
 823 
 824     *ipcs_shm = mainloop_add_ipc_server(PCMK__SERVER_BASED_SHM,
 825                                         QB_IPC_SHM, rw_cb);
 826 
 827     if (*ipcs_ro == NULL || *ipcs_rw == NULL || *ipcs_shm == NULL) {
 828         crm_err("Failed to create the CIB manager: exiting and inhibiting respawn");
 829         crm_warn("Verify pacemaker and pacemaker_remote are not both enabled");
 830         crm_exit(CRM_EX_FATAL);
 831     }
 832 }
 833 
 834 /*!
 835  * \internal
 836  * \brief Destroy IPC servers for pacemaker-based API
 837  *
 838  * \param[out] ipcs_ro   IPC server for read-only pacemaker-based API
 839  * \param[out] ipcs_rw   IPC server for read/write pacemaker-based API
 840  * \param[out] ipcs_shm  IPC server for shared-memory pacemaker-based API
 841  *
 842  * \note This is a convenience function for calling qb_ipcs_destroy() for each
 843  *       argument.
 844  */
 845 void
 846 pcmk__stop_based_ipc(qb_ipcs_service_t *ipcs_ro,
     /*  */
 847                      qb_ipcs_service_t *ipcs_rw,
 848                      qb_ipcs_service_t *ipcs_shm)
 849 {
 850     qb_ipcs_destroy(ipcs_ro);
 851     qb_ipcs_destroy(ipcs_rw);
 852     qb_ipcs_destroy(ipcs_shm);
 853 }
 854 
 855 /*!
 856  * \internal
 857  * \brief Add an IPC server to the main loop for the pacemaker-controld API
 858  *
 859  * \param[in] cb  IPC callbacks
 860  *
 861  * \return Newly created IPC server
 862  */
 863 qb_ipcs_service_t *
 864 pcmk__serve_controld_ipc(struct qb_ipcs_service_handlers *cb)
     /*  */
 865 {
 866     return mainloop_add_ipc_server(CRM_SYSTEM_CRMD, QB_IPC_NATIVE, cb);
 867 }
 868 
 869 /*!
 870  * \internal
 871  * \brief Add an IPC server to the main loop for the pacemaker-attrd API
 872  *
 873  * \param[in] cb  IPC callbacks
 874  *
 875  * \note This function exits fatally if unable to create the servers.
 876  */
 877 void
 878 pcmk__serve_attrd_ipc(qb_ipcs_service_t **ipcs,
     /*  */
 879                       struct qb_ipcs_service_handlers *cb)
 880 {
 881     *ipcs = mainloop_add_ipc_server(T_ATTRD, QB_IPC_NATIVE, cb);
 882 
 883     if (*ipcs == NULL) {
 884         crm_err("Failed to create pacemaker-attrd server: exiting and inhibiting respawn");
 885         crm_warn("Verify pacemaker and pacemaker_remote are not both enabled.");
 886         crm_exit(CRM_EX_FATAL);
 887     }
 888 }
 889 
 890 /*!
 891  * \internal
 892  * \brief Add an IPC server to the main loop for the pacemaker-fenced API
 893  *
 894  * \param[in] cb  IPC callbacks
 895  *
 896  * \note This function exits fatally if unable to create the servers.
 897  */
 898 void
 899 pcmk__serve_fenced_ipc(qb_ipcs_service_t **ipcs,
     /*  */
 900                        struct qb_ipcs_service_handlers *cb)
 901 {
 902     *ipcs = mainloop_add_ipc_server_with_prio("stonith-ng", QB_IPC_NATIVE, cb,
 903                                               QB_LOOP_HIGH);
 904 
 905     if (*ipcs == NULL) {
 906         crm_err("Failed to create fencer: exiting and inhibiting respawn.");
 907         crm_warn("Verify pacemaker and pacemaker_remote are not both enabled.");
 908         crm_exit(CRM_EX_FATAL);
 909     }
 910 }
 911 
 912 /*!
 913  * \brief Check whether string represents a client name used by cluster daemons
 914  *
 915  * \param[in] name  String to check
 916  *
 917  * \return true if name is standard client name used by daemons, false otherwise
 918  *
 919  * \note This is provided by the client, and so cannot be used by itself as a
 920  *       secure means of authentication.
 921  */
 922 bool
 923 crm_is_daemon_name(const char *name)
     /*  */
 924 {
 925     name = pcmk__message_name(name);
 926     return (!strcmp(name, CRM_SYSTEM_CRMD)
 927             || !strcmp(name, CRM_SYSTEM_STONITHD)
 928             || !strcmp(name, "stonith-ng")
 929             || !strcmp(name, "attrd")
 930             || !strcmp(name, CRM_SYSTEM_CIB)
 931             || !strcmp(name, CRM_SYSTEM_MCP)
 932             || !strcmp(name, CRM_SYSTEM_DC)
 933             || !strcmp(name, CRM_SYSTEM_TENGINE)
 934             || !strcmp(name, CRM_SYSTEM_LRMD));
 935 }
/* */
root/lib/common/ipc_server.c

DEFINITIONS
