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

DEFINITIONS
