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-2022 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         if (rc != BZ_OK) {
 425             crm_err("Decompression failed: %s " CRM_XS " bzerror=%d",
 426                     bz2_strerror(rc), rc);
 427             free(uncompressed);
 428             return NULL;
 429         }
 430     }
 431 
 432     CRM_ASSERT(text[header->size_uncompressed - 1] == 0);
 433 
 434     xml = string2xml(text);
 435     crm_log_xml_trace(xml, "[IPC received]");
 436 
 437     free(uncompressed);
 438     return xml;
 439 }
 440 
 441 static int crm_ipcs_flush_events(pcmk__client_t *c);
 442 
 443 static gboolean
 444 crm_ipcs_flush_events_cb(gpointer data)
     /*  */
 445 {
 446     pcmk__client_t *c = data;
 447 
 448     c->event_timer = 0;
 449     crm_ipcs_flush_events(c);
 450     return FALSE;
 451 }
 452 
 453 /*!
 454  * \internal
 455  * \brief Add progressive delay before next event queue flush
 456  *
 457  * \param[in,out] c          Client connection to add delay to
 458  * \param[in]     queue_len  Current event queue length
 459  */
 460 static inline void
 461 delay_next_flush(pcmk__client_t *c, unsigned int queue_len)
     /*  */
 462 {
 463     /* Delay a maximum of 1.5 seconds */
 464     guint delay = (queue_len < 5)? (1000 + 100 * queue_len) : 1500;
 465 
 466     c->event_timer = g_timeout_add(delay, crm_ipcs_flush_events_cb, c);
 467 }
 468 
 469 /*!
 470  * \internal
 471  * \brief Send client any messages in its queue
 472  *
 473  * \param[in,out] c  Client to flush
 474  *
 475  * \return Standard Pacemaker return value
 476  */
 477 static int
 478 crm_ipcs_flush_events(pcmk__client_t *c)
     /*  */
 479 {
 480     int rc = pcmk_rc_ok;
 481     ssize_t qb_rc = 0;
 482     unsigned int sent = 0;
 483     unsigned int queue_len = 0;
 484 
 485     if (c == NULL) {
 486         return rc;
 487 
 488     } else if (c->event_timer) {
 489         /* There is already a timer, wait until it goes off */
 490         crm_trace("Timer active for %p - %d", c->ipcs, c->event_timer);
 491         return rc;
 492     }
 493 
 494     if (c->event_queue) {
 495         queue_len = g_queue_get_length(c->event_queue);
 496     }
 497     while (sent < 100) {
 498         pcmk__ipc_header_t *header = NULL;
 499         struct iovec *event = NULL;
 500 
 501         if (c->event_queue) {
 502             // We don't pop unless send is successful
 503             event = g_queue_peek_head(c->event_queue);
 504         }
 505         if (event == NULL) { // Queue is empty
 506             break;
 507         }
 508 
 509         qb_rc = qb_ipcs_event_sendv(c->ipcs, event, 2);
 510         if (qb_rc < 0) {
 511             rc = (int) -qb_rc;
 512             break;
 513         }
 514         event = g_queue_pop_head(c->event_queue);
 515 
 516         sent++;
 517         header = event[0].iov_base;
 518         if (header->size_compressed) {
 519             crm_trace("Event %d to %p[%d] (%lld compressed bytes) sent",
 520                       header->qb.id, c->ipcs, c->pid, (long long) qb_rc);
 521         } else {
 522             crm_trace("Event %d to %p[%d] (%lld bytes) sent: %.120s",
 523                       header->qb.id, c->ipcs, c->pid, (long long) qb_rc,
 524                       (char *) (event[1].iov_base));
 525         }
 526         pcmk_free_ipc_event(event);
 527     }
 528 
 529     queue_len -= sent;
 530     if (sent > 0 || queue_len) {
 531         crm_trace("Sent %d events (%d remaining) for %p[%d]: %s (%lld)",
 532                   sent, queue_len, c->ipcs, c->pid,
 533                   pcmk_rc_str(rc), (long long) qb_rc);
 534     }
 535 
 536     if (queue_len) {
 537 
 538         /* Allow clients to briefly fall behind on processing incoming messages,
 539          * but drop completely unresponsive clients so the connection doesn't
 540          * consume resources indefinitely.
 541          */
 542         if (queue_len > QB_MAX(c->queue_max, PCMK_IPC_DEFAULT_QUEUE_MAX)) {
 543             if ((c->queue_backlog <= 1) || (queue_len < c->queue_backlog)) {
 544                 /* Don't evict for a new or shrinking backlog */
 545                 crm_warn("Client with process ID %u has a backlog of %u messages "
 546                          CRM_XS " %p", c->pid, queue_len, c->ipcs);
 547             } else {
 548                 crm_err("Evicting client with process ID %u due to backlog of %u messages "
 549                          CRM_XS " %p", c->pid, queue_len, c->ipcs);
 550                 c->queue_backlog = 0;
 551                 qb_ipcs_disconnect(c->ipcs);
 552                 return rc;
 553             }
 554         }
 555 
 556         c->queue_backlog = queue_len;
 557         delay_next_flush(c, queue_len);
 558 
 559     } else {
 560         /* Event queue is empty, there is no backlog */
 561         c->queue_backlog = 0;
 562     }
 563 
 564     return rc;
 565 }
 566 
 567 /*!
 568  * \internal
 569  * \brief Create an I/O vector for sending an IPC XML message
 570  *
 571  * \param[in]     request        Identifier for libqb response header
 572  * \param[in,out] message        XML message to send
 573  * \param[in]     max_send_size  If 0, default IPC buffer size is used
 574  * \param[out]    result         Where to store prepared I/O vector
 575  * \param[out]    bytes          Size of prepared data in bytes
 576  *
 577  * \return Standard Pacemaker return code
 578  */
 579 int
 580 pcmk__ipc_prepare_iov(uint32_t request, xmlNode *message,
     /*  */
 581                       uint32_t max_send_size, struct iovec **result,
 582                       ssize_t *bytes)
 583 {
 584     static unsigned int biggest = 0;
 585     struct iovec *iov;
 586     unsigned int total = 0;
 587     char *compressed = NULL;
 588     char *buffer = NULL;
 589     pcmk__ipc_header_t *header = NULL;
 590 
 591     if ((message == NULL) || (result == NULL)) {
 592         return EINVAL;
 593     }
 594 
 595     header = calloc(1, sizeof(pcmk__ipc_header_t));
 596     if (header == NULL) {
 597        return ENOMEM; /* errno mightn't be set by allocator */
 598     }
 599 
 600     buffer = dump_xml_unformatted(message);
 601 
 602     if (max_send_size == 0) {
 603         max_send_size = crm_ipc_default_buffer_size();
 604     }
 605     CRM_LOG_ASSERT(max_send_size != 0);
 606 
 607     *result = NULL;
 608     iov = pcmk__new_ipc_event();
 609     iov[0].iov_len = sizeof(pcmk__ipc_header_t);
 610     iov[0].iov_base = header;
 611 
 612     header->version = PCMK__IPC_VERSION;
 613     header->size_uncompressed = 1 + strlen(buffer);
 614     total = iov[0].iov_len + header->size_uncompressed;
 615 
 616     if (total < max_send_size) {
 617         iov[1].iov_base = buffer;
 618         iov[1].iov_len = header->size_uncompressed;
 619 
 620     } else {
 621         unsigned int new_size = 0;
 622 
 623         if (pcmk__compress(buffer, (unsigned int) header->size_uncompressed,
 624                            (unsigned int) max_send_size, &compressed,
 625                            &new_size) == pcmk_rc_ok) {
 626 
 627             pcmk__set_ipc_flags(header->flags, "send data", crm_ipc_compressed);
 628             header->size_compressed = new_size;
 629 
 630             iov[1].iov_len = header->size_compressed;
 631             iov[1].iov_base = compressed;
 632 
 633             free(buffer);
 634 
 635             biggest = QB_MAX(header->size_compressed, biggest);
 636 
 637         } else {
 638             crm_log_xml_trace(message, "EMSGSIZE");
 639             biggest = QB_MAX(header->size_uncompressed, biggest);
 640 
 641             crm_err("Could not compress %u-byte message into less than IPC "
 642                     "limit of %u bytes; set PCMK_ipc_buffer to higher value "
 643                     "(%u bytes suggested)",
 644                     header->size_uncompressed, max_send_size, 4 * biggest);
 645 
 646             free(compressed);
 647             free(buffer);
 648             pcmk_free_ipc_event(iov);
 649             return EMSGSIZE;
 650         }
 651     }
 652 
 653     header->qb.size = iov[0].iov_len + iov[1].iov_len;
 654     header->qb.id = (int32_t)request;    /* Replying to a specific request */
 655 
 656     *result = iov;
 657     CRM_ASSERT(header->qb.size > 0);
 658     if (bytes != NULL) {
 659         *bytes = header->qb.size;
 660     }
 661     return pcmk_rc_ok;
 662 }
 663 
 664 int
 665 pcmk__ipc_send_iov(pcmk__client_t *c, struct iovec *iov, uint32_t flags)
     /*  */
 666 {
 667     int rc = pcmk_rc_ok;
 668     static uint32_t id = 1;
 669     pcmk__ipc_header_t *header = iov[0].iov_base;
 670 
 671     if (c->flags & pcmk__client_proxied) {
 672         /* _ALL_ replies to proxied connections need to be sent as events */
 673         if (!pcmk_is_set(flags, crm_ipc_server_event)) {
 674             /* The proxied flag lets us know this was originally meant to be a
 675              * response, even though we're sending it over the event channel.
 676              */
 677             pcmk__set_ipc_flags(flags, "server event",
 678                                 crm_ipc_server_event
 679                                 |crm_ipc_proxied_relay_response);
 680         }
 681     }
 682 
 683     pcmk__set_ipc_flags(header->flags, "server event", flags);
 684     if (flags & crm_ipc_server_event) {
 685         header->qb.id = id++;   /* We don't really use it, but doesn't hurt to set one */
 686 
 687         if (flags & crm_ipc_server_free) {
 688             crm_trace("Sending the original to %p[%d]", c->ipcs, c->pid);
 689             add_event(c, iov);
 690 
 691         } else {
 692             struct iovec *iov_copy = pcmk__new_ipc_event();
 693 
 694             crm_trace("Sending a copy to %p[%d]", c->ipcs, c->pid);
 695             iov_copy[0].iov_len = iov[0].iov_len;
 696             iov_copy[0].iov_base = malloc(iov[0].iov_len);
 697             memcpy(iov_copy[0].iov_base, iov[0].iov_base, iov[0].iov_len);
 698 
 699             iov_copy[1].iov_len = iov[1].iov_len;
 700             iov_copy[1].iov_base = malloc(iov[1].iov_len);
 701             memcpy(iov_copy[1].iov_base, iov[1].iov_base, iov[1].iov_len);
 702 
 703             add_event(c, iov_copy);
 704         }
 705 
 706     } else {
 707         ssize_t qb_rc;
 708 
 709         CRM_LOG_ASSERT(header->qb.id != 0);     /* Replying to a specific request */
 710 
 711         qb_rc = qb_ipcs_response_sendv(c->ipcs, iov, 2);
 712         if (qb_rc < header->qb.size) {
 713             if (qb_rc < 0) {
 714                 rc = (int) -qb_rc;
 715             }
 716             crm_notice("Response %d to pid %d failed: %s "
 717                        CRM_XS " bytes=%u rc=%lld ipcs=%p",
 718                        header->qb.id, c->pid, pcmk_rc_str(rc),
 719                        header->qb.size, (long long) qb_rc, c->ipcs);
 720 
 721         } else {
 722             crm_trace("Response %d sent, %lld bytes to %p[%d]",
 723                       header->qb.id, (long long) qb_rc, c->ipcs, c->pid);
 724         }
 725 
 726         if (flags & crm_ipc_server_free) {
 727             pcmk_free_ipc_event(iov);
 728         }
 729     }
 730 
 731     if (flags & crm_ipc_server_event) {
 732         rc = crm_ipcs_flush_events(c);
 733     } else {
 734         crm_ipcs_flush_events(c);
 735     }
 736 
 737     if ((rc == EPIPE) || (rc == ENOTCONN)) {
 738         crm_trace("Client %p disconnected", c->ipcs);
 739     }
 740     return rc;
 741 }
 742 
 743 int
 744 pcmk__ipc_send_xml(pcmk__client_t *c, uint32_t request, xmlNode *message,
     /*  */
 745                    uint32_t flags)
 746 {
 747     struct iovec *iov = NULL;
 748     int rc = pcmk_rc_ok;
 749 
 750     if (c == NULL) {
 751         return EINVAL;
 752     }
 753     rc = pcmk__ipc_prepare_iov(request, message, crm_ipc_default_buffer_size(),
 754                                &iov, NULL);
 755     if (rc == pcmk_rc_ok) {
 756         pcmk__set_ipc_flags(flags, "send data", crm_ipc_server_free);
 757         rc = pcmk__ipc_send_iov(c, iov, flags);
 758     } else {
 759         pcmk_free_ipc_event(iov);
 760         crm_notice("IPC message to pid %d failed: %s " CRM_XS " rc=%d",
 761                    c->pid, pcmk_rc_str(rc), rc);
 762     }
 763     return rc;
 764 }
 765 
 766 /*!
 767  * \internal
 768  * \brief Create an acknowledgement with a status code to send to a client
 769  *
 770  * \param[in] function  Calling function
 771  * \param[in] line      Source file line within calling function
 772  * \param[in] flags     IPC flags to use when sending
 773  * \param[in] tag       Element name to use for acknowledgement
 774  * \param[in] ver       IPC protocol version (can be NULL)
 775  * \param[in] status    Exit status code to add to ack
 776  *
 777  * \return Newly created XML for ack
 778  * \note The caller is responsible for freeing the return value with free_xml().
 779  */
 780 xmlNode *
 781 pcmk__ipc_create_ack_as(const char *function, int line, uint32_t flags,
     /*  */
 782                         const char *tag, const char *ver, crm_exit_t status)
 783 {
 784     xmlNode *ack = NULL;
 785 
 786     if (pcmk_is_set(flags, crm_ipc_client_response)) {
 787         ack = create_xml_node(NULL, tag);
 788         crm_xml_add(ack, "function", function);
 789         crm_xml_add_int(ack, "line", line);
 790         crm_xml_add_int(ack, "status", (int) status);
 791         crm_xml_add(ack, PCMK__XA_IPC_PROTO_VERSION, ver);
 792     }
 793     return ack;
 794 }
 795 
 796 /*!
 797  * \internal
 798  * \brief Send an acknowledgement with a status code to a client
 799  *
 800  * \param[in] function  Calling function
 801  * \param[in] line      Source file line within calling function
 802  * \param[in] c         Client to send ack to
 803  * \param[in] request   Request ID being replied to
 804  * \param[in] flags     IPC flags to use when sending
 805  * \param[in] tag       Element name to use for acknowledgement
 806  * \param[in] ver       IPC protocol version (can be NULL)
 807  * \param[in] status    Status code to send with acknowledgement
 808  *
 809  * \return Standard Pacemaker return code
 810  */
 811 int
 812 pcmk__ipc_send_ack_as(const char *function, int line, pcmk__client_t *c,
     /*  */
 813                       uint32_t request, uint32_t flags, const char *tag,
 814                       const char *ver, crm_exit_t status)
 815 {
 816     int rc = pcmk_rc_ok;
 817     xmlNode *ack = pcmk__ipc_create_ack_as(function, line, flags, tag, ver, status);
 818 
 819     if (ack != NULL) {
 820         crm_trace("Ack'ing IPC message from client %s as <%s status=%d>",
 821                   pcmk__client_name(c), tag, status);
 822         c->request_id = 0;
 823         rc = pcmk__ipc_send_xml(c, request, ack, flags);
 824         free_xml(ack);
 825     }
 826     return rc;
 827 }
 828 
 829 /*!
 830  * \internal
 831  * \brief Add an IPC server to the main loop for the pacemaker-based API
 832  *
 833  * \param[out] ipcs_ro   New IPC server for read-only pacemaker-based API
 834  * \param[out] ipcs_rw   New IPC server for read/write pacemaker-based API
 835  * \param[out] ipcs_shm  New IPC server for shared-memory pacemaker-based API
 836  * \param[in]  ro_cb     IPC callbacks for read-only API
 837  * \param[in]  rw_cb     IPC callbacks for read/write and shared-memory APIs
 838  *
 839  * \note This function exits fatally if unable to create the servers.
 840  */
 841 void pcmk__serve_based_ipc(qb_ipcs_service_t **ipcs_ro,
     /*  */
 842                            qb_ipcs_service_t **ipcs_rw,
 843                            qb_ipcs_service_t **ipcs_shm,
 844                            struct qb_ipcs_service_handlers *ro_cb,
 845                            struct qb_ipcs_service_handlers *rw_cb)
 846 {
 847     *ipcs_ro = mainloop_add_ipc_server(PCMK__SERVER_BASED_RO,
 848                                        QB_IPC_NATIVE, ro_cb);
 849 
 850     *ipcs_rw = mainloop_add_ipc_server(PCMK__SERVER_BASED_RW,
 851                                        QB_IPC_NATIVE, rw_cb);
 852 
 853     *ipcs_shm = mainloop_add_ipc_server(PCMK__SERVER_BASED_SHM,
 854                                         QB_IPC_SHM, rw_cb);
 855 
 856     if (*ipcs_ro == NULL || *ipcs_rw == NULL || *ipcs_shm == NULL) {
 857         crm_err("Failed to create the CIB manager: exiting and inhibiting respawn");
 858         crm_warn("Verify pacemaker and pacemaker_remote are not both enabled");
 859         crm_exit(CRM_EX_FATAL);
 860     }
 861 }
 862 
 863 /*!
 864  * \internal
 865  * \brief Destroy IPC servers for pacemaker-based API
 866  *
 867  * \param[out] ipcs_ro   IPC server for read-only pacemaker-based API
 868  * \param[out] ipcs_rw   IPC server for read/write pacemaker-based API
 869  * \param[out] ipcs_shm  IPC server for shared-memory pacemaker-based API
 870  *
 871  * \note This is a convenience function for calling qb_ipcs_destroy() for each
 872  *       argument.
 873  */
 874 void
 875 pcmk__stop_based_ipc(qb_ipcs_service_t *ipcs_ro,
     /*  */
 876                      qb_ipcs_service_t *ipcs_rw,
 877                      qb_ipcs_service_t *ipcs_shm)
 878 {
 879     qb_ipcs_destroy(ipcs_ro);
 880     qb_ipcs_destroy(ipcs_rw);
 881     qb_ipcs_destroy(ipcs_shm);
 882 }
 883 
 884 /*!
 885  * \internal
 886  * \brief Add an IPC server to the main loop for the pacemaker-controld API
 887  *
 888  * \param[in] cb  IPC callbacks
 889  *
 890  * \return Newly created IPC server
 891  */
 892 qb_ipcs_service_t *
 893 pcmk__serve_controld_ipc(struct qb_ipcs_service_handlers *cb)
     /*  */
 894 {
 895     return mainloop_add_ipc_server(CRM_SYSTEM_CRMD, QB_IPC_NATIVE, cb);
 896 }
 897 
 898 /*!
 899  * \internal
 900  * \brief Add an IPC server to the main loop for the pacemaker-attrd API
 901  *
 902  * \param[out] ipcs  Where to store newly created IPC server
 903  * \param[in] cb  IPC callbacks
 904  *
 905  * \note This function exits fatally if unable to create the servers.
 906  */
 907 void
 908 pcmk__serve_attrd_ipc(qb_ipcs_service_t **ipcs,
     /*  */
 909                       struct qb_ipcs_service_handlers *cb)
 910 {
 911     *ipcs = mainloop_add_ipc_server(T_ATTRD, QB_IPC_NATIVE, cb);
 912 
 913     if (*ipcs == NULL) {
 914         crm_err("Failed to create pacemaker-attrd server: exiting and inhibiting respawn");
 915         crm_warn("Verify pacemaker and pacemaker_remote are not both enabled.");
 916         crm_exit(CRM_EX_FATAL);
 917     }
 918 }
 919 
 920 /*!
 921  * \internal
 922  * \brief Add an IPC server to the main loop for the pacemaker-fenced API
 923  *
 924  * \param[out] ipcs  Where to store newly created IPC server
 925  * \param[in]  cb    IPC callbacks
 926  *
 927  * \note This function exits fatally if unable to create the servers.
 928  */
 929 void
 930 pcmk__serve_fenced_ipc(qb_ipcs_service_t **ipcs,
     /*  */
 931                        struct qb_ipcs_service_handlers *cb)
 932 {
 933     *ipcs = mainloop_add_ipc_server_with_prio("stonith-ng", QB_IPC_NATIVE, cb,
 934                                               QB_LOOP_HIGH);
 935 
 936     if (*ipcs == NULL) {
 937         crm_err("Failed to create fencer: exiting and inhibiting respawn.");
 938         crm_warn("Verify pacemaker and pacemaker_remote are not both enabled.");
 939         crm_exit(CRM_EX_FATAL);
 940     }
 941 }
 942 
 943 /*!
 944  * \internal
 945  * \brief Add an IPC server to the main loop for the pacemakerd API
 946  *
 947  * \param[out] ipcs  Where to store newly created IPC server
 948  * \param[in]  cb    IPC callbacks
 949  *
 950  * \note This function exits with CRM_EX_OSERR if unable to create the servers.
 951  */
 952 void
 953 pcmk__serve_pacemakerd_ipc(qb_ipcs_service_t **ipcs,
     /*  */
 954                        struct qb_ipcs_service_handlers *cb)
 955 {
 956     *ipcs = mainloop_add_ipc_server(CRM_SYSTEM_MCP, QB_IPC_NATIVE, cb);
 957 
 958     if (*ipcs == NULL) {
 959         crm_err("Couldn't start pacemakerd IPC server");
 960         crm_warn("Verify pacemaker and pacemaker_remote are not both enabled.");
 961         /* sub-daemons are observed by pacemakerd. Thus we exit CRM_EX_FATAL
 962          * if we want to prevent pacemakerd from restarting them.
 963          * With pacemakerd we leave the exit-code shown to e.g. systemd
 964          * to what it was prior to moving the code here from pacemakerd.c
 965          */
 966         crm_exit(CRM_EX_OSERR);
 967     }
 968 }
 969 
 970 /*!
 971  * \internal
 972  * \brief Add an IPC server to the main loop for the pacemaker-schedulerd API
 973  *
 974  * \param[in] cb  IPC callbacks
 975  *
 976  * \return Newly created IPC server
 977  * \note This function exits fatally if unable to create the servers.
 978  */
 979 qb_ipcs_service_t *
 980 pcmk__serve_schedulerd_ipc(struct qb_ipcs_service_handlers *cb)
     /*  */
 981 {
 982     return mainloop_add_ipc_server(CRM_SYSTEM_PENGINE, QB_IPC_NATIVE, cb);
 983 }
 984 
 985 /*!
 986  * \brief Check whether string represents a client name used by cluster daemons
 987  *
 988  * \param[in] name  String to check
 989  *
 990  * \return true if name is standard client name used by daemons, false otherwise
 991  *
 992  * \note This is provided by the client, and so cannot be used by itself as a
 993  *       secure means of authentication.
 994  */
 995 bool
 996 crm_is_daemon_name(const char *name)
     /*  */
 997 {
 998     name = pcmk__message_name(name);
 999     return (!strcmp(name, CRM_SYSTEM_CRMD)
1000             || !strcmp(name, CRM_SYSTEM_STONITHD)
1001             || !strcmp(name, "stonith-ng")
1002             || !strcmp(name, "attrd")
1003             || !strcmp(name, CRM_SYSTEM_CIB)
1004             || !strcmp(name, CRM_SYSTEM_MCP)
1005             || !strcmp(name, CRM_SYSTEM_DC)
1006             || !strcmp(name, CRM_SYSTEM_TENGINE)
1007             || !strcmp(name, CRM_SYSTEM_LRMD));
1008 }
/* */
root/lib/common/ipc_server.c

DEFINITIONS
