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

DEFINITIONS
