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

DEFINITIONS
