src/bin/lttng-sessiond/consumer.cpp

   1 /*
   2  * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
   3  * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * SPDX-License-Identifier: GPL-2.0-only
   6  *
   7  */
   8
   9 #define _LGPL_SOURCE
  10 #include "consumer.hpp"
  11 #include "health-sessiond.hpp"
  12 #include "lttng-sessiond.hpp"
  13 #include "ust-app.hpp"
  14 #include "utils.hpp"
  15
  16 #include <common/common.hpp>
  17 #include <common/defaults.hpp>
  18 #include <common/relayd/relayd.hpp>
  19 #include <common/string-utils/format.hpp>
  20 #include <common/urcu.hpp>
  21 #include <common/uri.hpp>
  22
  23 #include <inttypes.h>
  24 #include <stdio.h>
  25 #include <stdlib.h>
  26 #include <string.h>
  27 #include <sys/stat.h>
  28 #include <sys/types.h>
  29 #include <unistd.h>
  30
  31 /*
  32  * Return allocated full pathname of the session using the consumer trace path
  33  * and subdir if available.
  34  *
  35  * The caller can safely free(3) the returned value. On error, NULL is
  36  * returned.
  37  */
  38 char *setup_channel_trace_path(struct consumer_output *consumer,
  39                                const char *session_path,
  40                                size_t *consumer_path_offset)
  41 {
  42         int ret;
  43         char *pathname;
  44
  45         LTTNG_ASSERT(consumer);
  46         LTTNG_ASSERT(session_path);
  47
  48         health_code_update();
  49
  50         /*
  51          * Allocate the string ourself to make sure we never exceed
  52          * LTTNG_PATH_MAX.
  53          */
  54         pathname = calloc<char>(LTTNG_PATH_MAX);
  55         if (!pathname) {
  56                 goto error;
  57         }
  58
  59         /* Get correct path name destination */
  60         if (consumer->type == CONSUMER_DST_NET && consumer->relay_major_version == 2 &&
  61             consumer->relay_minor_version < 11) {
  62                 ret = snprintf(pathname,
  63                                LTTNG_PATH_MAX,
  64                                "%s%s/%s/%s",
  65                                consumer->dst.net.base_dir,
  66                                consumer->chunk_path,
  67                                consumer->domain_subdir,
  68                                session_path);
  69                 *consumer_path_offset = 0;
  70         } else {
  71                 ret = snprintf(
  72                         pathname, LTTNG_PATH_MAX, "%s/%s", consumer->domain_subdir, session_path);
  73                 *consumer_path_offset = strlen(consumer->domain_subdir) + 1;
  74         }
  75         DBG3("Consumer trace path relative to current trace chunk: \"%s\"", pathname);
  76         if (ret < 0) {
  77                 PERROR("Failed to format channel path");
  78                 goto error;
  79         } else if (ret >= LTTNG_PATH_MAX) {
  80                 ERR("Truncation occurred while formatting channel path");
  81                 goto error;
  82         }
  83
  84         return pathname;
  85 error:
  86         free(pathname);
  87         return nullptr;
  88 }
  89
  90 /*
  91  * Send a data payload using a given consumer socket of size len.
  92  *
  93  * The consumer socket lock MUST be acquired before calling this since this
  94  * function can change the fd value.
  95  *
  96  * Return 0 on success else a negative value on error.
  97  */
  98 int consumer_socket_send(struct consumer_socket *socket, const void *msg, size_t len)
  99 {
 100         int fd;
 101         ssize_t size;
 102
 103         LTTNG_ASSERT(socket);
 104         LTTNG_ASSERT(socket->fd_ptr);
 105         LTTNG_ASSERT(msg);
 106
 107         /* Consumer socket is invalid. Stopping. */
 108         fd = *socket->fd_ptr;
 109         if (fd < 0) {
 110                 goto error;
 111         }
 112
 113         size = lttcomm_send_unix_sock(fd, msg, len);
 114         if (size < 0) {
 115                 /* The above call will print a PERROR on error. */
 116                 DBG("Error when sending data to consumer on sock %d", fd);
 117                 /*
 118                  * At this point, the socket is not usable anymore thus closing it and
 119                  * setting the file descriptor to -1 so it is not reused.
 120                  */
 121
 122                 /* This call will PERROR on error. */
 123                 (void) lttcomm_close_unix_sock(fd);
 124                 *socket->fd_ptr = -1;
 125                 goto error;
 126         }
 127
 128         return 0;
 129
 130 error:
 131         return -1;
 132 }
 133
 134 /*
 135  * Receive a data payload using a given consumer socket of size len.
 136  *
 137  * The consumer socket lock MUST be acquired before calling this since this
 138  * function can change the fd value.
 139  *
 140  * Return 0 on success else a negative value on error.
 141  */
 142 int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
 143 {
 144         int fd;
 145         ssize_t size;
 146
 147         LTTNG_ASSERT(socket);
 148         LTTNG_ASSERT(socket->fd_ptr);
 149         LTTNG_ASSERT(msg);
 150
 151         /* Consumer socket is invalid. Stopping. */
 152         fd = *socket->fd_ptr;
 153         if (fd < 0) {
 154                 goto error;
 155         }
 156
 157         size = lttcomm_recv_unix_sock(fd, msg, len);
 158         if (size <= 0) {
 159                 /* The above call will print a PERROR on error. */
 160                 DBG("Error when receiving data from the consumer socket %d", fd);
 161                 /*
 162                  * At this point, the socket is not usable anymore thus closing it and
 163                  * setting the file descriptor to -1 so it is not reused.
 164                  */
 165
 166                 /* This call will PERROR on error. */
 167                 (void) lttcomm_close_unix_sock(fd);
 168                 *socket->fd_ptr = -1;
 169                 goto error;
 170         }
 171
 172         return 0;
 173
 174 error:
 175         return -1;
 176 }
 177
 178 /*
 179  * Receive a reply command status message from the consumer. Consumer socket
 180  * lock MUST be acquired before calling this function.
 181  *
 182  * Return 0 on success, -1 on recv error or a negative lttng error code which
 183  * was possibly returned by the consumer.
 184  */
 185 int consumer_recv_status_reply(struct consumer_socket *sock)
 186 {
 187         int ret;
 188         struct lttcomm_consumer_status_msg reply;
 189
 190         LTTNG_ASSERT(sock);
 191
 192         ret = consumer_socket_recv(sock, &reply, sizeof(reply));
 193         if (ret < 0) {
 194                 goto end;
 195         }
 196
 197         if (reply.ret_code == LTTCOMM_CONSUMERD_SUCCESS) {
 198                 /* All good. */
 199                 ret = 0;
 200         } else {
 201                 ret = -reply.ret_code;
 202                 DBG("Consumer ret code %d", ret);
 203         }
 204
 205 end:
 206         return ret;
 207 }
 208
 209 /*
 210  * Once the ASK_CHANNEL command is sent to the consumer, the channel
 211  * information are sent back. This call receives that data and populates key
 212  * and stream_count.
 213  *
 214  * On success return 0 and both key and stream_count are set. On error, a
 215  * negative value is sent back and both parameters are untouched.
 216  */
 217 int consumer_recv_status_channel(struct consumer_socket *sock,
 218                                  uint64_t *key,
 219                                  unsigned int *stream_count)
 220 {
 221         int ret;
 222         struct lttcomm_consumer_status_channel reply;
 223
 224         LTTNG_ASSERT(sock);
 225         LTTNG_ASSERT(stream_count);
 226         LTTNG_ASSERT(key);
 227
 228         ret = consumer_socket_recv(sock, &reply, sizeof(reply));
 229         if (ret < 0) {
 230                 goto end;
 231         }
 232
 233         /* An error is possible so don't touch the key and stream_count. */
 234         if (reply.ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
 235                 ret = -1;
 236                 goto end;
 237         }
 238
 239         *key = reply.key;
 240         *stream_count = reply.stream_count;
 241         ret = 0;
 242
 243 end:
 244         return ret;
 245 }
 246
 247 /*
 248  * Send destroy relayd command to consumer.
 249  *
 250  * On success return positive value. On error, negative value.
 251  */
 252 int consumer_send_destroy_relayd(struct consumer_socket *sock, struct consumer_output *consumer)
 253 {
 254         int ret;
 255         struct lttcomm_consumer_msg msg;
 256
 257         LTTNG_ASSERT(consumer);
 258         LTTNG_ASSERT(sock);
 259
 260         DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
 261
 262         memset(&msg, 0, sizeof(msg));
 263         msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
 264         msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
 265
 266         pthread_mutex_lock(sock->lock);
 267         ret = consumer_socket_send(sock, &msg, sizeof(msg));
 268         if (ret < 0) {
 269                 goto error;
 270         }
 271
 272         /* Don't check the return value. The caller will do it. */
 273         ret = consumer_recv_status_reply(sock);
 274
 275         DBG2("Consumer send destroy relayd command done");
 276
 277 error:
 278         pthread_mutex_unlock(sock->lock);
 279         return ret;
 280 }
 281
 282 /*
 283  * For each consumer socket in the consumer output object, send a destroy
 284  * relayd command.
 285  */
 286 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
 287 {
 288         struct lttng_ht_iter iter;
 289         struct consumer_socket *socket;
 290
 291         LTTNG_ASSERT(consumer);
 292
 293         /* Destroy any relayd connection */
 294         if (consumer->type == CONSUMER_DST_NET) {
 295                 lttng::urcu::read_lock_guard read_lock;
 296
 297                 cds_lfht_for_each_entry (consumer->socks->ht, &iter.iter, socket, node.node) {
 298                         /* Send destroy relayd command. */
 299                         const int ret = consumer_send_destroy_relayd(socket, consumer);
 300
 301                         if (ret < 0) {
 302                                 DBG("Unable to send destroy relayd command to consumer");
 303                                 /* Continue since we MUST delete everything at this point. */
 304                         }
 305                 }
 306         }
 307 }
 308
 309 /*
 310  * From a consumer_data structure, allocate and add a consumer socket to the
 311  * consumer output.
 312  *
 313  * Return 0 on success, else negative value on error
 314  */
 315 int consumer_create_socket(struct consumer_data *data, struct consumer_output *output)
 316 {
 317         int ret = 0;
 318         struct consumer_socket *socket;
 319
 320         LTTNG_ASSERT(data);
 321
 322         lttng::urcu::read_lock_guard read_lock;
 323
 324         if (output == nullptr || data->cmd_sock < 0) {
 325                 /*
 326                  * Not an error. Possible there is simply not spawned consumer or it's
 327                  * disabled for the tracing session asking the socket.
 328                  */
 329                 goto error;
 330         }
 331
 332         socket = consumer_find_socket(data->cmd_sock, output);
 333         if (socket == nullptr) {
 334                 socket = consumer_allocate_socket(&data->cmd_sock);
 335                 if (socket == nullptr) {
 336                         ret = -1;
 337                         goto error;
 338                 }
 339
 340                 socket->registered = 0;
 341                 socket->lock = &data->lock;
 342                 consumer_add_socket(socket, output);
 343         }
 344
 345         socket->type = data->type;
 346
 347         DBG3("Consumer socket created (fd: %d) and added to output", data->cmd_sock);
 348
 349 error:
 350         return ret;
 351 }
 352
 353 /*
 354  * Return the consumer socket from the given consumer output with the right
 355  * bitness. On error, returns NULL.
 356  *
 357  * The caller MUST acquire a rcu read side lock and keep it until the socket
 358  * object reference is not needed anymore.
 359  */
 360 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
 361                                                         const struct consumer_output *consumer)
 362 {
 363         int consumer_fd;
 364         struct consumer_socket *socket = nullptr;
 365
 366         ASSERT_RCU_READ_LOCKED();
 367
 368         switch (bits) {
 369         case 64:
 370                 consumer_fd = uatomic_read(&the_ust_consumerd64_fd);
 371                 break;
 372         case 32:
 373                 consumer_fd = uatomic_read(&the_ust_consumerd32_fd);
 374                 break;
 375         default:
 376                 abort();
 377                 goto end;
 378         }
 379
 380         socket = consumer_find_socket(consumer_fd, consumer);
 381         if (!socket) {
 382                 ERR("Consumer socket fd %d not found in consumer obj %p", consumer_fd, consumer);
 383         }
 384
 385 end:
 386         return socket;
 387 }
 388
 389 /*
 390  * Find a consumer_socket in a consumer_output hashtable. Read side lock must
 391  * be acquired before calling this function and across use of the
 392  * returned consumer_socket.
 393  */
 394 struct consumer_socket *consumer_find_socket(int key, const struct consumer_output *consumer)
 395 {
 396         struct lttng_ht_iter iter;
 397         struct lttng_ht_node_ulong *node;
 398         struct consumer_socket *socket = nullptr;
 399
 400         ASSERT_RCU_READ_LOCKED();
 401
 402         /* Negative keys are lookup failures */
 403         if (key < 0 || consumer == nullptr) {
 404                 return nullptr;
 405         }
 406
 407         lttng_ht_lookup(consumer->socks, (void *) ((unsigned long) key), &iter);
 408         node = lttng_ht_iter_get_node_ulong(&iter);
 409         if (node != nullptr) {
 410                 socket = lttng::utils::container_of(node, &consumer_socket::node);
 411         }
 412
 413         return socket;
 414 }
 415
 416 /*
 417  * Allocate a new consumer_socket and return the pointer.
 418  */
 419 struct consumer_socket *consumer_allocate_socket(int *fd)
 420 {
 421         struct consumer_socket *socket = nullptr;
 422
 423         LTTNG_ASSERT(fd);
 424
 425         socket = zmalloc<consumer_socket>();
 426         if (socket == nullptr) {
 427                 PERROR("zmalloc consumer socket");
 428                 goto error;
 429         }
 430
 431         socket->fd_ptr = fd;
 432         lttng_ht_node_init_ulong(&socket->node, *fd);
 433
 434 error:
 435         return socket;
 436 }
 437
 438 /*
 439  * Add consumer socket to consumer output object. Read side lock must be
 440  * acquired before calling this function.
 441  */
 442 void consumer_add_socket(struct consumer_socket *sock, struct consumer_output *consumer)
 443 {
 444         LTTNG_ASSERT(sock);
 445         LTTNG_ASSERT(consumer);
 446         ASSERT_RCU_READ_LOCKED();
 447
 448         lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
 449 }
 450
 451 /*
 452  * Delete consumer socket to consumer output object. Read side lock must be
 453  * acquired before calling this function.
 454  */
 455 void consumer_del_socket(struct consumer_socket *sock, struct consumer_output *consumer)
 456 {
 457         int ret;
 458         struct lttng_ht_iter iter;
 459
 460         LTTNG_ASSERT(sock);
 461         LTTNG_ASSERT(consumer);
 462         ASSERT_RCU_READ_LOCKED();
 463
 464         iter.iter.node = &sock->node.node;
 465         ret = lttng_ht_del(consumer->socks, &iter);
 466         LTTNG_ASSERT(!ret);
 467 }
 468
 469 /*
 470  * RCU destroy call function.
 471  */
 472 static void destroy_socket_rcu(struct rcu_head *head)
 473 {
 474         struct lttng_ht_node_ulong *node =
 475                 lttng::utils::container_of(head, &lttng_ht_node_ulong::head);
 476         struct consumer_socket *socket = lttng::utils::container_of(node, &consumer_socket::node);
 477
 478         free(socket);
 479 }
 480
 481 /*
 482  * Destroy and free socket pointer in a call RCU. The call must either:
 483  *  - have acquired the read side lock before calling this function, or
 484  *  - guarantee the validity of the `struct consumer_socket` object for the
 485  *    duration of the call.
 486  */
 487 void consumer_destroy_socket(struct consumer_socket *sock)
 488 {
 489         LTTNG_ASSERT(sock);
 490
 491         /*
 492          * We DO NOT close the file descriptor here since it is global to the
 493          * session daemon and is closed only if the consumer dies or a custom
 494          * consumer was registered,
 495          */
 496         if (sock->registered) {
 497                 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
 498                 lttcomm_close_unix_sock(*sock->fd_ptr);
 499         }
 500
 501         call_rcu(&sock->node.head, destroy_socket_rcu);
 502 }
 503
 504 /*
 505  * Allocate and assign data to a consumer_output object.
 506  *
 507  * Return pointer to structure.
 508  */
 509 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
 510 {
 511         struct consumer_output *output = nullptr;
 512
 513         output = zmalloc<consumer_output>();
 514         if (output == nullptr) {
 515                 PERROR("zmalloc consumer_output");
 516                 goto error;
 517         }
 518
 519         /* By default, consumer output is enabled */
 520         output->enabled = true;
 521         output->type = type;
 522         output->net_seq_index = (uint64_t) -1ULL;
 523         urcu_ref_init(&output->ref);
 524
 525         output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 526
 527 error:
 528         return output;
 529 }
 530
 531 /*
 532  * Iterate over the consumer output socket hash table and destroy them. The
 533  * socket file descriptor are only closed if the consumer output was
 534  * registered meaning it's an external consumer.
 535  */
 536 void consumer_destroy_output_sockets(struct consumer_output *obj)
 537 {
 538         struct lttng_ht_iter iter;
 539         struct consumer_socket *socket;
 540
 541         if (!obj->socks) {
 542                 return;
 543         }
 544
 545         {
 546                 lttng::urcu::read_lock_guard read_lock;
 547
 548                 cds_lfht_for_each_entry (obj->socks->ht, &iter.iter, socket, node.node) {
 549                         consumer_del_socket(socket, obj);
 550                         consumer_destroy_socket(socket);
 551                 }
 552         }
 553 }
 554
 555 /*
 556  * Delete the consumer_output object from the list and free the ptr.
 557  */
 558 static void consumer_release_output(struct urcu_ref *ref)
 559 {
 560         struct consumer_output *obj = lttng::utils::container_of(ref, &consumer_output::ref);
 561
 562         consumer_destroy_output_sockets(obj);
 563
 564         if (obj->socks) {
 565                 /* Finally destroy HT */
 566                 lttng_ht_destroy(obj->socks);
 567         }
 568
 569         free(obj);
 570 }
 571
 572 /*
 573  * Get the consumer_output object.
 574  */
 575 void consumer_output_get(struct consumer_output *obj)
 576 {
 577         urcu_ref_get(&obj->ref);
 578 }
 579
 580 /*
 581  * Put the consumer_output object.
 582  */
 583 void consumer_output_put(struct consumer_output *obj)
 584 {
 585         if (!obj) {
 586                 return;
 587         }
 588         urcu_ref_put(&obj->ref, consumer_release_output);
 589 }
 590
 591 /*
 592  * Copy consumer output and returned the newly allocated copy.
 593  */
 594 struct consumer_output *consumer_copy_output(struct consumer_output *src)
 595 {
 596         int ret;
 597         struct consumer_output *output;
 598
 599         LTTNG_ASSERT(src);
 600
 601         output = consumer_create_output(src->type);
 602         if (output == nullptr) {
 603                 goto end;
 604         }
 605         output->enabled = src->enabled;
 606         output->net_seq_index = src->net_seq_index;
 607         memcpy(output->domain_subdir, src->domain_subdir, sizeof(output->domain_subdir));
 608         output->snapshot = src->snapshot;
 609         output->relay_major_version = src->relay_major_version;
 610         output->relay_minor_version = src->relay_minor_version;
 611         output->relay_allows_clear = src->relay_allows_clear;
 612         memcpy(&output->dst, &src->dst, sizeof(output->dst));
 613         ret = consumer_copy_sockets(output, src);
 614         if (ret < 0) {
 615                 goto error_put;
 616         }
 617 end:
 618         return output;
 619
 620 error_put:
 621         consumer_output_put(output);
 622         return nullptr;
 623 }
 624
 625 /*
 626  * Copy consumer sockets from src to dst.
 627  *
 628  * Return 0 on success or else a negative value.
 629  */
 630 int consumer_copy_sockets(struct consumer_output *dst, struct consumer_output *src)
 631 {
 632         int ret = 0;
 633         struct lttng_ht_iter iter;
 634         struct consumer_socket *socket, *copy_sock;
 635
 636         LTTNG_ASSERT(dst);
 637         LTTNG_ASSERT(src);
 638
 639         {
 640                 lttng::urcu::read_lock_guard read_lock;
 641
 642                 cds_lfht_for_each_entry (src->socks->ht, &iter.iter, socket, node.node) {
 643                         /* Ignore socket that are already there. */
 644                         copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
 645                         if (copy_sock) {
 646                                 continue;
 647                         }
 648
 649                         /* Create new socket object. */
 650                         copy_sock = consumer_allocate_socket(socket->fd_ptr);
 651                         if (copy_sock == nullptr) {
 652                                 ret = -ENOMEM;
 653                                 goto error;
 654                         }
 655
 656                         copy_sock->registered = socket->registered;
 657                         /*
 658                          * This is valid because this lock is shared accross all consumer
 659                          * object being the global lock of the consumer data structure of the
 660                          * session daemon.
 661                          */
 662                         copy_sock->lock = socket->lock;
 663                         consumer_add_socket(copy_sock, dst);
 664                 }
 665         }
 666
 667 error:
 668         return ret;
 669 }
 670
 671 /*
 672  * Set network URI to the consumer output.
 673  *
 674  * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
 675  * error.
 676  */
 677 int consumer_set_network_uri(const struct ltt_session *session,
 678                              struct consumer_output *output,
 679                              struct lttng_uri *uri)
 680 {
 681         int ret;
 682         struct lttng_uri *dst_uri = nullptr;
 683
 684         /* Code flow error safety net. */
 685         LTTNG_ASSERT(output);
 686         LTTNG_ASSERT(uri);
 687
 688         switch (uri->stype) {
 689         case LTTNG_STREAM_CONTROL:
 690                 dst_uri = &output->dst.net.control;
 691                 output->dst.net.control_isset = 1;
 692                 if (uri->port == 0) {
 693                         /* Assign default port. */
 694                         uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 695                 } else {
 696                         if (output->dst.net.data_isset && uri->port == output->dst.net.data.port) {
 697                                 ret = -LTTNG_ERR_INVALID;
 698                                 goto error;
 699                         }
 700                 }
 701                 DBG3("Consumer control URI set with port %d", uri->port);
 702                 break;
 703         case LTTNG_STREAM_DATA:
 704                 dst_uri = &output->dst.net.data;
 705                 output->dst.net.data_isset = 1;
 706                 if (uri->port == 0) {
 707                         /* Assign default port. */
 708                         uri->port = DEFAULT_NETWORK_DATA_PORT;
 709                 } else {
 710                         if (output->dst.net.control_isset &&
 711                             uri->port == output->dst.net.control.port) {
 712                                 ret = -LTTNG_ERR_INVALID;
 713                                 goto error;
 714                         }
 715                 }
 716                 DBG3("Consumer data URI set with port %d", uri->port);
 717                 break;
 718         default:
 719                 ERR("Set network uri type unknown %d", uri->stype);
 720                 ret = -LTTNG_ERR_INVALID;
 721                 goto error;
 722         }
 723
 724         ret = uri_compare(dst_uri, uri);
 725         if (!ret) {
 726                 /* Same URI, don't touch it and return success. */
 727                 DBG3("URI network compare are the same");
 728                 goto equal;
 729         }
 730
 731         /* URIs were not equal, replacing it. */
 732         memcpy(dst_uri, uri, sizeof(struct lttng_uri));
 733         output->type = CONSUMER_DST_NET;
 734         if (dst_uri->stype != LTTNG_STREAM_CONTROL) {
 735                 /* Only the control uri needs to contain the path. */
 736                 goto end;
 737         }
 738
 739         /*
 740          * If the user has specified a subdir as part of the control
 741          * URL, the session's base output directory is:
 742          *   /RELAYD_OUTPUT_PATH/HOSTNAME/USER_SPECIFIED_DIR
 743          *
 744          * Hence, the "base_dir" from which all stream files and
 745          * session rotation chunks are created takes the form
 746          *   /HOSTNAME/USER_SPECIFIED_DIR
 747          *
 748          * If the user has not specified an output directory as part of
 749          * the control URL, the base output directory has the form:
 750          *   /RELAYD_OUTPUT_PATH/HOSTNAME/SESSION_NAME-CREATION_TIME
 751          *
 752          * Hence, the "base_dir" from which all stream files and
 753          * session rotation chunks are created takes the form
 754          *   /HOSTNAME/SESSION_NAME-CREATION_TIME
 755          *
 756          * Note that automatically generated session names already
 757          * contain the session's creation time. In that case, the
 758          * creation time is omitted to prevent it from being duplicated
 759          * in the final directory hierarchy.
 760          */
 761         if (*uri->subdir) {
 762                 if (strstr(uri->subdir, "../")) {
 763                         ERR("Network URI subdirs are not allowed to walk up the path hierarchy");
 764                         ret = -LTTNG_ERR_INVALID;
 765                         goto error;
 766                 }
 767                 ret = snprintf(output->dst.net.base_dir,
 768                                sizeof(output->dst.net.base_dir),
 769                                "/%s/%s/",
 770                                session->hostname,
 771                                uri->subdir);
 772         } else {
 773                 if (session->has_auto_generated_name) {
 774                         ret = snprintf(output->dst.net.base_dir,
 775                                        sizeof(output->dst.net.base_dir),
 776                                        "/%s/%s/",
 777                                        session->hostname,
 778                                        session->name);
 779                 } else {
 780                         char session_creation_datetime[16];
 781                         size_t strftime_ret;
 782                         struct tm *timeinfo;
 783
 784                         timeinfo = localtime(&session->creation_time);
 785                         if (!timeinfo) {
 786                                 ret = -LTTNG_ERR_FATAL;
 787                                 goto error;
 788                         }
 789                         strftime_ret = strftime(session_creation_datetime,
 790                                                 sizeof(session_creation_datetime),
 791                                                 "%Y%m%d-%H%M%S",
 792                                                 timeinfo);
 793                         if (strftime_ret == 0) {
 794                                 ERR("Failed to format session creation timestamp while setting network URI");
 795                                 ret = -LTTNG_ERR_FATAL;
 796                                 goto error;
 797                         }
 798                         ret = snprintf(output->dst.net.base_dir,
 799                                        sizeof(output->dst.net.base_dir),
 800                                        "/%s/%s-%s/",
 801                                        session->hostname,
 802                                        session->name,
 803                                        session_creation_datetime);
 804                 }
 805         }
 806         if (ret >= sizeof(output->dst.net.base_dir)) {
 807                 ret = -LTTNG_ERR_INVALID;
 808                 ERR("Truncation occurred while setting network output base directory");
 809                 goto error;
 810         } else if (ret == -1) {
 811                 ret = -LTTNG_ERR_INVALID;
 812                 PERROR("Error occurred while setting network output base directory");
 813                 goto error;
 814         }
 815
 816         DBG3("Consumer set network uri base_dir path %s", output->dst.net.base_dir);
 817
 818 end:
 819         return 0;
 820 equal:
 821         return 1;
 822 error:
 823         return ret;
 824 }
 825
 826 /*
 827  * Send file descriptor to consumer via sock.
 828  *
 829  * The consumer socket lock must be held by the caller.
 830  */
 831 int consumer_send_fds(struct consumer_socket *sock, const int *fds, size_t nb_fd)
 832 {
 833         int ret;
 834
 835         LTTNG_ASSERT(fds);
 836         LTTNG_ASSERT(sock);
 837         LTTNG_ASSERT(nb_fd > 0);
 838         LTTNG_ASSERT(pthread_mutex_trylock(sock->lock) == EBUSY);
 839
 840         ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
 841         if (ret < 0) {
 842                 /* The above call will print a PERROR on error. */
 843                 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
 844                 goto error;
 845         }
 846
 847         ret = consumer_recv_status_reply(sock);
 848 error:
 849         return ret;
 850 }
 851
 852 /*
 853  * Consumer send communication message structure to consumer.
 854  *
 855  * The consumer socket lock must be held by the caller.
 856  */
 857 int consumer_send_msg(struct consumer_socket *sock, const struct lttcomm_consumer_msg *msg)
 858 {
 859         int ret;
 860
 861         LTTNG_ASSERT(msg);
 862         LTTNG_ASSERT(sock);
 863         LTTNG_ASSERT(pthread_mutex_trylock(sock->lock) == EBUSY);
 864
 865         ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
 866         if (ret < 0) {
 867                 goto error;
 868         }
 869
 870         ret = consumer_recv_status_reply(sock);
 871
 872 error:
 873         return ret;
 874 }
 875
 876 /*
 877  * Consumer send channel communication message structure to consumer.
 878  *
 879  * The consumer socket lock must be held by the caller.
 880  */
 881 int consumer_send_channel(struct consumer_socket *sock, struct lttcomm_consumer_msg *msg)
 882 {
 883         int ret;
 884
 885         LTTNG_ASSERT(msg);
 886         LTTNG_ASSERT(sock);
 887
 888         ret = consumer_send_msg(sock, msg);
 889         if (ret < 0) {
 890                 goto error;
 891         }
 892
 893 error:
 894         return ret;
 895 }
 896
 897 /*
 898  * Populate the given consumer msg structure with the ask_channel command
 899  * information.
 900  */
 901 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 902                                         uint64_t subbuf_size,
 903                                         uint64_t num_subbuf,
 904                                         int overwrite,
 905                                         unsigned int switch_timer_interval,
 906                                         unsigned int read_timer_interval,
 907                                         unsigned int live_timer_interval,
 908                                         bool is_in_live_session,
 909                                         unsigned int monitor_timer_interval,
 910                                         int output,
 911                                         int type,
 912                                         uint64_t session_id,
 913                                         const char *pathname,
 914                                         const char *name,
 915                                         uint64_t relayd_id,
 916                                         uint64_t key,
 917                                         const lttng_uuid& uuid,
 918                                         uint32_t chan_id,
 919                                         uint64_t tracefile_size,
 920                                         uint64_t tracefile_count,
 921                                         uint64_t session_id_per_pid,
 922                                         unsigned int monitor,
 923                                         uint32_t ust_app_uid,
 924                                         int64_t blocking_timeout,
 925                                         const char *root_shm_path,
 926                                         const char *shm_path,
 927                                         struct lttng_trace_chunk *trace_chunk,
 928                                         const struct lttng_credentials *buffer_credentials)
 929 {
 930         LTTNG_ASSERT(msg);
 931
 932         /* Zeroed structure */
 933         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 934         msg->u.ask_channel.buffer_credentials.uid = UINT32_MAX;
 935         msg->u.ask_channel.buffer_credentials.gid = UINT32_MAX;
 936
 937         if (trace_chunk) {
 938                 uint64_t chunk_id;
 939                 enum lttng_trace_chunk_status chunk_status;
 940
 941                 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
 942                 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 943                 LTTNG_OPTIONAL_SET(&msg->u.ask_channel.chunk_id, chunk_id);
 944         }
 945         msg->u.ask_channel.buffer_credentials.uid = lttng_credentials_get_uid(buffer_credentials);
 946         msg->u.ask_channel.buffer_credentials.gid = lttng_credentials_get_gid(buffer_credentials);
 947
 948         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 949         msg->u.ask_channel.subbuf_size = subbuf_size;
 950         msg->u.ask_channel.num_subbuf = num_subbuf;
 951         msg->u.ask_channel.overwrite = overwrite;
 952         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 953         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 954         msg->u.ask_channel.live_timer_interval = live_timer_interval;
 955         msg->u.ask_channel.is_live = is_in_live_session;
 956         msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
 957         msg->u.ask_channel.output = output;
 958         msg->u.ask_channel.type = type;
 959         msg->u.ask_channel.session_id = session_id;
 960         msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
 961         msg->u.ask_channel.relayd_id = relayd_id;
 962         msg->u.ask_channel.key = key;
 963         msg->u.ask_channel.chan_id = chan_id;
 964         msg->u.ask_channel.tracefile_size = tracefile_size;
 965         msg->u.ask_channel.tracefile_count = tracefile_count;
 966         msg->u.ask_channel.monitor = monitor;
 967         msg->u.ask_channel.ust_app_uid = ust_app_uid;
 968         msg->u.ask_channel.blocking_timeout = blocking_timeout;
 969
 970         std::copy(uuid.begin(), uuid.end(), msg->u.ask_channel.uuid);
 971
 972         if (pathname) {
 973                 strncpy(msg->u.ask_channel.pathname, pathname, sizeof(msg->u.ask_channel.pathname));
 974                 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname) - 1] = '\0';
 975         }
 976
 977         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 978         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 979
 980         if (root_shm_path) {
 981                 strncpy(msg->u.ask_channel.root_shm_path,
 982                         root_shm_path,
 983                         sizeof(msg->u.ask_channel.root_shm_path));
 984                 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] =
 985                         '\0';
 986         }
 987         if (shm_path) {
 988                 strncpy(msg->u.ask_channel.shm_path, shm_path, sizeof(msg->u.ask_channel.shm_path));
 989                 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
 990         }
 991 }
 992
 993 /*
 994  * Init channel communication message structure.
 995  */
 996 void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 997                                         uint64_t channel_key,
 998                                         uint64_t session_id,
 999                                         const char *pathname,
1000                                         uint64_t relayd_id,
1001                                         const char *name,
1002                                         unsigned int nb_init_streams,
1003                                         enum lttng_event_output output,
1004                                         int type,
1005                                         uint64_t tracefile_size,
1006                                         uint64_t tracefile_count,
1007                                         unsigned int monitor,
1008                                         unsigned int live_timer_interval,
1009                                         bool is_in_live_session,
1010                                         unsigned int monitor_timer_interval,
1011                                         struct lttng_trace_chunk *trace_chunk)
1012 {
1013         LTTNG_ASSERT(msg);
1014
1015         /* Zeroed structure */
1016         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1017
1018         if (trace_chunk) {
1019                 uint64_t chunk_id;
1020                 enum lttng_trace_chunk_status chunk_status;
1021
1022                 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
1023                 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
1024                 LTTNG_OPTIONAL_SET(&msg->u.channel.chunk_id, chunk_id);
1025         }
1026
1027         /* Send channel */
1028         msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
1029         msg->u.channel.channel_key = channel_key;
1030         msg->u.channel.session_id = session_id;
1031         msg->u.channel.relayd_id = relayd_id;
1032         msg->u.channel.nb_init_streams = nb_init_streams;
1033         msg->u.channel.output = output;
1034         msg->u.channel.type = type;
1035         msg->u.channel.tracefile_size = tracefile_size;
1036         msg->u.channel.tracefile_count = tracefile_count;
1037         msg->u.channel.monitor = monitor;
1038         msg->u.channel.live_timer_interval = live_timer_interval;
1039         msg->u.channel.is_live = is_in_live_session;
1040         msg->u.channel.monitor_timer_interval = monitor_timer_interval;
1041
1042         strncpy(msg->u.channel.pathname, pathname, sizeof(msg->u.channel.pathname));
1043         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
1044
1045         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
1046         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
1047 }
1048
1049 /*
1050  * Init stream communication message structure.
1051  */
1052 void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg,
1053                                        uint64_t channel_key,
1054                                        uint64_t stream_key,
1055                                        int32_t cpu)
1056 {
1057         LTTNG_ASSERT(msg);
1058
1059         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1060
1061         msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM;
1062         msg->u.stream.channel_key = channel_key;
1063         msg->u.stream.stream_key = stream_key;
1064         msg->u.stream.cpu = cpu;
1065 }
1066
1067 void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
1068                                          enum lttng_consumer_command cmd,
1069                                          uint64_t channel_key,
1070                                          uint64_t net_seq_idx)
1071 {
1072         LTTNG_ASSERT(msg);
1073
1074         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1075
1076         msg->cmd_type = cmd;
1077         msg->u.sent_streams.channel_key = channel_key;
1078         msg->u.sent_streams.net_seq_idx = net_seq_idx;
1079 }
1080
1081 /*
1082  * Send stream communication structure to the consumer.
1083  */
1084 int consumer_send_stream(struct consumer_socket *sock,
1085                          struct consumer_output *dst,
1086                          struct lttcomm_consumer_msg *msg,
1087                          const int *fds,
1088                          size_t nb_fd)
1089 {
1090         int ret;
1091
1092         LTTNG_ASSERT(msg);
1093         LTTNG_ASSERT(dst);
1094         LTTNG_ASSERT(sock);
1095         LTTNG_ASSERT(fds);
1096
1097         ret = consumer_send_msg(sock, msg);
1098         if (ret < 0) {
1099                 goto error;
1100         }
1101
1102         ret = consumer_send_fds(sock, fds, nb_fd);
1103         if (ret < 0) {
1104                 goto error;
1105         }
1106
1107 error:
1108         return ret;
1109 }
1110
1111 /*
1112  * Send relayd socket to consumer associated with a session name.
1113  *
1114  * The consumer socket lock must be held by the caller.
1115  *
1116  * On success return positive value. On error, negative value.
1117  */
1118 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
1119                                 struct lttcomm_relayd_sock *rsock,
1120                                 struct consumer_output *consumer,
1121                                 enum lttng_stream_type type,
1122                                 uint64_t session_id,
1123                                 const char *session_name,
1124                                 const char *hostname,
1125                                 const char *base_path,
1126                                 int session_live_timer,
1127                                 const uint64_t *current_chunk_id,
1128                                 time_t session_creation_time,
1129                                 bool session_name_contains_creation_time)
1130 {
1131         int ret;
1132         int fd;
1133         struct lttcomm_consumer_msg msg;
1134
1135         /* Code flow error. Safety net. */
1136         LTTNG_ASSERT(rsock);
1137         LTTNG_ASSERT(consumer);
1138         LTTNG_ASSERT(consumer_sock);
1139
1140         memset(&msg, 0, sizeof(msg));
1141         /* Bail out if consumer is disabled */
1142         if (!consumer->enabled) {
1143                 ret = LTTNG_OK;
1144                 goto error;
1145         }
1146
1147         if (type == LTTNG_STREAM_CONTROL) {
1148                 char output_path[LTTNG_PATH_MAX] = {};
1149                 uint64_t relayd_session_id;
1150
1151                 ret = relayd_create_session(rsock,
1152                                             &relayd_session_id,
1153                                             session_name,
1154                                             hostname,
1155                                             base_path,
1156                                             session_live_timer,
1157                                             consumer->snapshot,
1158                                             session_id,
1159                                             the_sessiond_uuid,
1160                                             current_chunk_id,
1161                                             session_creation_time,
1162                                             session_name_contains_creation_time,
1163                                             output_path);
1164                 if (ret < 0) {
1165                         /* Close the control socket. */
1166                         (void) relayd_close(rsock);
1167                         goto error;
1168                 }
1169                 msg.u.relayd_sock.relayd_session_id = relayd_session_id;
1170                 DBG("Created session on relay, output path reply: %s", output_path);
1171         }
1172
1173         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1174         /*
1175          * Assign network consumer output index using the temporary consumer since
1176          * this call should only be made from within a set_consumer_uri() function
1177          * call in the session daemon.
1178          */
1179         msg.u.relayd_sock.net_index = consumer->net_seq_index;
1180         msg.u.relayd_sock.type = type;
1181         msg.u.relayd_sock.session_id = session_id;
1182         msg.u.relayd_sock.major = rsock->major;
1183         msg.u.relayd_sock.minor = rsock->minor;
1184         msg.u.relayd_sock.relayd_socket_protocol = rsock->sock.proto;
1185
1186         DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
1187         ret = consumer_send_msg(consumer_sock, &msg);
1188         if (ret < 0) {
1189                 goto error;
1190         }
1191
1192         DBG3("Sending relayd socket file descriptor to consumer");
1193         fd = rsock->sock.fd;
1194         ret = consumer_send_fds(consumer_sock, &fd, 1);
1195         if (ret < 0) {
1196                 goto error;
1197         }
1198
1199         DBG2("Consumer relayd socket sent");
1200
1201 error:
1202         return ret;
1203 }
1204
1205 static int
1206 consumer_send_pipe(struct consumer_socket *consumer_sock, enum lttng_consumer_command cmd, int pipe)
1207 {
1208         int ret;
1209         struct lttcomm_consumer_msg msg;
1210         const char *pipe_name;
1211         const char *command_name;
1212
1213         switch (cmd) {
1214         case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1215                 pipe_name = "channel monitor";
1216                 command_name = "SET_CHANNEL_MONITOR_PIPE";
1217                 break;
1218         default:
1219                 ERR("Unexpected command received in %s (cmd = %d)", __func__, (int) cmd);
1220                 abort();
1221         }
1222
1223         /* Code flow error. Safety net. */
1224
1225         memset(&msg, 0, sizeof(msg));
1226         msg.cmd_type = cmd;
1227
1228         pthread_mutex_lock(consumer_sock->lock);
1229         DBG3("Sending %s command to consumer", command_name);
1230         ret = consumer_send_msg(consumer_sock, &msg);
1231         if (ret < 0) {
1232                 goto error;
1233         }
1234
1235         DBG3("Sending %s pipe %d to consumer on socket %d", pipe_name, pipe, *consumer_sock->fd_ptr);
1236         ret = consumer_send_fds(consumer_sock, &pipe, 1);
1237         if (ret < 0) {
1238                 goto error;
1239         }
1240
1241         DBG2("%s pipe successfully sent", pipe_name);
1242 error:
1243         pthread_mutex_unlock(consumer_sock->lock);
1244         return ret;
1245 }
1246
1247 int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock, int pipe)
1248 {
1249         return consumer_send_pipe(consumer_sock, LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe);
1250 }
1251
1252 /*
1253  * Ask the consumer if the data is pending for the specific session id.
1254  * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1255  */
1256 int consumer_is_data_pending(uint64_t session_id, struct consumer_output *consumer)
1257 {
1258         int ret;
1259         int32_t ret_code = 0; /* Default is that the data is NOT pending */
1260         struct consumer_socket *socket;
1261         struct lttng_ht_iter iter;
1262         struct lttcomm_consumer_msg msg;
1263
1264         LTTNG_ASSERT(consumer);
1265
1266         DBG3("Consumer data pending for id %" PRIu64, session_id);
1267
1268         memset(&msg, 0, sizeof(msg));
1269         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1270         msg.u.data_pending.session_id = session_id;
1271
1272         {
1273                 /* Send command for each consumer. */
1274                 lttng::urcu::read_lock_guard read_lock;
1275
1276                 cds_lfht_for_each_entry (consumer->socks->ht, &iter.iter, socket, node.node) {
1277                         pthread_mutex_lock(socket->lock);
1278                         ret = consumer_socket_send(socket, &msg, sizeof(msg));
1279                         if (ret < 0) {
1280                                 pthread_mutex_unlock(socket->lock);
1281                                 goto error_unlock;
1282                         }
1283
1284                         /*
1285                          * No need for a recv reply status because the answer to the command is
1286                          * the reply status message.
1287                          */
1288                         ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1289                         if (ret < 0) {
1290                                 pthread_mutex_unlock(socket->lock);
1291                                 goto error_unlock;
1292                         }
1293
1294                         pthread_mutex_unlock(socket->lock);
1295
1296                         if (ret_code == 1) {
1297                                 break;
1298                         }
1299                 }
1300         }
1301
1302         DBG("Consumer data is %s pending for session id %" PRIu64,
1303             ret_code == 1 ? "" : "NOT",
1304             session_id);
1305         return ret_code;
1306
1307 error_unlock:
1308         return -1;
1309 }
1310
1311 /*
1312  * Send a flush command to consumer using the given channel key.
1313  *
1314  * Return 0 on success else a negative value.
1315  */
1316 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1317 {
1318         int ret;
1319         struct lttcomm_consumer_msg msg;
1320
1321         LTTNG_ASSERT(socket);
1322
1323         DBG2("Consumer flush channel key %" PRIu64, key);
1324
1325         memset(&msg, 0, sizeof(msg));
1326         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1327         msg.u.flush_channel.key = key;
1328
1329         pthread_mutex_lock(socket->lock);
1330         health_code_update();
1331
1332         ret = consumer_send_msg(socket, &msg);
1333         if (ret < 0) {
1334                 goto end;
1335         }
1336
1337 end:
1338         health_code_update();
1339         pthread_mutex_unlock(socket->lock);
1340         return ret;
1341 }
1342
1343 /*
1344  * Send a clear quiescent command to consumer using the given channel key.
1345  *
1346  * Return 0 on success else a negative value.
1347  */
1348 int consumer_clear_quiescent_channel(struct consumer_socket *socket, uint64_t key)
1349 {
1350         int ret;
1351         struct lttcomm_consumer_msg msg;
1352
1353         LTTNG_ASSERT(socket);
1354
1355         DBG2("Consumer clear quiescent channel key %" PRIu64, key);
1356
1357         memset(&msg, 0, sizeof(msg));
1358         msg.cmd_type = LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL;
1359         msg.u.clear_quiescent_channel.key = key;
1360
1361         pthread_mutex_lock(socket->lock);
1362         health_code_update();
1363
1364         ret = consumer_send_msg(socket, &msg);
1365         if (ret < 0) {
1366                 goto end;
1367         }
1368
1369 end:
1370         health_code_update();
1371         pthread_mutex_unlock(socket->lock);
1372         return ret;
1373 }
1374
1375 /*
1376  * Send a close metadata command to consumer using the given channel key.
1377  * Called with registry lock held.
1378  *
1379  * Return 0 on success else a negative value.
1380  */
1381 int consumer_close_metadata(struct consumer_socket *socket, uint64_t metadata_key)
1382 {
1383         int ret;
1384         struct lttcomm_consumer_msg msg;
1385
1386         LTTNG_ASSERT(socket);
1387
1388         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1389
1390         memset(&msg, 0, sizeof(msg));
1391         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1392         msg.u.close_metadata.key = metadata_key;
1393
1394         pthread_mutex_lock(socket->lock);
1395         health_code_update();
1396
1397         ret = consumer_send_msg(socket, &msg);
1398         if (ret < 0) {
1399                 goto end;
1400         }
1401
1402 end:
1403         health_code_update();
1404         pthread_mutex_unlock(socket->lock);
1405         return ret;
1406 }
1407
1408 /*
1409  * Send a setup metdata command to consumer using the given channel key.
1410  *
1411  * Return 0 on success else a negative value.
1412  */
1413 int consumer_setup_metadata(struct consumer_socket *socket, uint64_t metadata_key)
1414 {
1415         int ret;
1416         struct lttcomm_consumer_msg msg;
1417
1418         LTTNG_ASSERT(socket);
1419
1420         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1421
1422         memset(&msg, 0, sizeof(msg));
1423         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1424         msg.u.setup_metadata.key = metadata_key;
1425
1426         pthread_mutex_lock(socket->lock);
1427         health_code_update();
1428
1429         ret = consumer_send_msg(socket, &msg);
1430         if (ret < 0) {
1431                 goto end;
1432         }
1433
1434 end:
1435         health_code_update();
1436         pthread_mutex_unlock(socket->lock);
1437         return ret;
1438 }
1439
1440 /*
1441  * Send metadata string to consumer.
1442  * RCU read-side lock must be held to guarantee existence of socket.
1443  *
1444  * Return 0 on success else a negative value.
1445  */
1446 int consumer_push_metadata(struct consumer_socket *socket,
1447                            uint64_t metadata_key,
1448                            char *metadata_str,
1449                            size_t len,
1450                            size_t target_offset,
1451                            uint64_t version)
1452 {
1453         int ret;
1454         struct lttcomm_consumer_msg msg;
1455
1456         LTTNG_ASSERT(socket);
1457         ASSERT_RCU_READ_LOCKED();
1458
1459         DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1460
1461         pthread_mutex_lock(socket->lock);
1462
1463         memset(&msg, 0, sizeof(msg));
1464         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1465         msg.u.push_metadata.key = metadata_key;
1466         msg.u.push_metadata.target_offset = target_offset;
1467         msg.u.push_metadata.len = len;
1468         msg.u.push_metadata.version = version;
1469
1470         health_code_update();
1471         ret = consumer_send_msg(socket, &msg);
1472         if (ret < 0 || len == 0) {
1473                 goto end;
1474         }
1475
1476         DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr, len);
1477
1478         ret = consumer_socket_send(socket, metadata_str, len);
1479         if (ret < 0) {
1480                 goto end;
1481         }
1482
1483         health_code_update();
1484         ret = consumer_recv_status_reply(socket);
1485         if (ret < 0) {
1486                 goto end;
1487         }
1488
1489 end:
1490         pthread_mutex_unlock(socket->lock);
1491         health_code_update();
1492         return ret;
1493 }
1494
1495 /*
1496  * Ask the consumer to snapshot a specific channel using the key.
1497  *
1498  * Returns LTTNG_OK on success or else an LTTng error code.
1499  */
1500 enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket,
1501                                                 uint64_t key,
1502                                                 const struct consumer_output *output,
1503                                                 int metadata,
1504                                                 const char *channel_path,
1505                                                 uint64_t nb_packets_per_stream)
1506 {
1507         int ret;
1508         enum lttng_error_code status = LTTNG_OK;
1509         struct lttcomm_consumer_msg msg;
1510
1511         LTTNG_ASSERT(socket);
1512         LTTNG_ASSERT(output);
1513
1514         DBG("Consumer snapshot channel key %" PRIu64, key);
1515
1516         memset(&msg, 0, sizeof(msg));
1517         msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1518         msg.u.snapshot_channel.key = key;
1519         msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
1520         msg.u.snapshot_channel.metadata = metadata;
1521
1522         if (output->type == CONSUMER_DST_NET) {
1523                 msg.u.snapshot_channel.relayd_id = output->net_seq_index;
1524                 msg.u.snapshot_channel.use_relayd = 1;
1525         } else {
1526                 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1527         }
1528         ret = lttng_strncpy(msg.u.snapshot_channel.pathname,
1529                             channel_path,
1530                             sizeof(msg.u.snapshot_channel.pathname));
1531         if (ret < 0) {
1532                 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1533                     sizeof(msg.u.snapshot_channel.pathname),
1534                     strlen(channel_path),
1535                     channel_path);
1536                 status = LTTNG_ERR_SNAPSHOT_FAIL;
1537                 goto error;
1538         }
1539
1540         health_code_update();
1541         pthread_mutex_lock(socket->lock);
1542         ret = consumer_send_msg(socket, &msg);
1543         pthread_mutex_unlock(socket->lock);
1544         if (ret < 0) {
1545                 switch (-ret) {
1546                 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1547                         status = LTTNG_ERR_CHAN_NOT_FOUND;
1548                         break;
1549                 default:
1550                         status = LTTNG_ERR_SNAPSHOT_FAIL;
1551                         break;
1552                 }
1553                 goto error;
1554         }
1555
1556 error:
1557         health_code_update();
1558         return status;
1559 }
1560
1561 /*
1562  * Ask the consumer the number of discarded events for a channel.
1563  */
1564 int consumer_get_discarded_events(uint64_t session_id,
1565                                   uint64_t channel_key,
1566                                   struct consumer_output *consumer,
1567                                   uint64_t *discarded)
1568 {
1569         int ret;
1570         struct consumer_socket *socket;
1571         struct lttng_ht_iter iter;
1572         struct lttcomm_consumer_msg msg;
1573
1574         LTTNG_ASSERT(consumer);
1575
1576         DBG3("Consumer discarded events id %" PRIu64, session_id);
1577
1578         memset(&msg, 0, sizeof(msg));
1579         msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1580         msg.u.discarded_events.session_id = session_id;
1581         msg.u.discarded_events.channel_key = channel_key;
1582
1583         *discarded = 0;
1584
1585         /* Send command for each consumer. */
1586         {
1587                 lttng::urcu::read_lock_guard read_lock;
1588
1589                 cds_lfht_for_each_entry (consumer->socks->ht, &iter.iter, socket, node.node) {
1590                         uint64_t consumer_discarded = 0;
1591
1592                         pthread_mutex_lock(socket->lock);
1593                         ret = consumer_socket_send(socket, &msg, sizeof(msg));
1594                         if (ret < 0) {
1595                                 pthread_mutex_unlock(socket->lock);
1596                                 goto end;
1597                         }
1598
1599                         /*
1600                          * No need for a recv reply status because the answer to the
1601                          * command is the reply status message.
1602                          */
1603                         ret = consumer_socket_recv(
1604                                 socket, &consumer_discarded, sizeof(consumer_discarded));
1605                         if (ret < 0) {
1606                                 ERR("get discarded events");
1607                                 pthread_mutex_unlock(socket->lock);
1608                                 goto end;
1609                         }
1610
1611                         pthread_mutex_unlock(socket->lock);
1612                         *discarded += consumer_discarded;
1613                 }
1614         }
1615
1616         ret = 0;
1617         DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64, *discarded, session_id);
1618
1619 end:
1620         return ret;
1621 }
1622
1623 /*
1624  * Ask the consumer the number of lost packets for a channel.
1625  */
1626 int consumer_get_lost_packets(uint64_t session_id,
1627                               uint64_t channel_key,
1628                               struct consumer_output *consumer,
1629                               uint64_t *lost)
1630 {
1631         int ret;
1632         struct consumer_socket *socket;
1633         struct lttng_ht_iter iter;
1634         struct lttcomm_consumer_msg msg;
1635
1636         LTTNG_ASSERT(consumer);
1637
1638         DBG3("Consumer lost packets id %" PRIu64, session_id);
1639
1640         memset(&msg, 0, sizeof(msg));
1641         msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1642         msg.u.lost_packets.session_id = session_id;
1643         msg.u.lost_packets.channel_key = channel_key;
1644
1645         *lost = 0;
1646
1647         /* Send command for each consumer. */
1648         {
1649                 lttng::urcu::read_lock_guard read_lock;
1650
1651                 cds_lfht_for_each_entry (consumer->socks->ht, &iter.iter, socket, node.node) {
1652                         uint64_t consumer_lost = 0;
1653                         pthread_mutex_lock(socket->lock);
1654                         ret = consumer_socket_send(socket, &msg, sizeof(msg));
1655                         if (ret < 0) {
1656                                 pthread_mutex_unlock(socket->lock);
1657                                 goto end;
1658                         }
1659
1660                         /*
1661                          * No need for a recv reply status because the answer to the
1662                          * command is the reply status message.
1663                          */
1664                         ret = consumer_socket_recv(socket, &consumer_lost, sizeof(consumer_lost));
1665                         if (ret < 0) {
1666                                 ERR("get lost packets");
1667                                 pthread_mutex_unlock(socket->lock);
1668                                 goto end;
1669                         }
1670                         pthread_mutex_unlock(socket->lock);
1671                         *lost += consumer_lost;
1672                 }
1673         }
1674
1675         ret = 0;
1676         DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64, *lost, session_id);
1677
1678 end:
1679         return ret;
1680 }
1681
1682 /*
1683  * Ask the consumer to rotate a channel.
1684  *
1685  * The new_chunk_id is the session->rotate_count that has been incremented
1686  * when the rotation started. On the relay, this allows to keep track in which
1687  * chunk each stream is currently writing to (for the rotate_pending operation).
1688  */
1689 int consumer_rotate_channel(struct consumer_socket *socket,
1690                             uint64_t key,
1691                             struct consumer_output *output,
1692                             bool is_metadata_channel)
1693 {
1694         int ret;
1695         struct lttcomm_consumer_msg msg;
1696
1697         LTTNG_ASSERT(socket);
1698
1699         DBG("Consumer rotate channel key %" PRIu64, key);
1700
1701         pthread_mutex_lock(socket->lock);
1702         memset(&msg, 0, sizeof(msg));
1703         msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL;
1704         msg.u.rotate_channel.key = key;
1705         msg.u.rotate_channel.metadata = !!is_metadata_channel;
1706
1707         if (output->type == CONSUMER_DST_NET) {
1708                 msg.u.rotate_channel.relayd_id = output->net_seq_index;
1709         } else {
1710                 msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL;
1711         }
1712
1713         health_code_update();
1714         ret = consumer_send_msg(socket, &msg);
1715         if (ret < 0) {
1716                 switch (-ret) {
1717                 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1718                         ret = -LTTNG_ERR_CHAN_NOT_FOUND;
1719                         break;
1720                 default:
1721                         ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER;
1722                         break;
1723                 }
1724                 goto error;
1725         }
1726 error:
1727         pthread_mutex_unlock(socket->lock);
1728         health_code_update();
1729         return ret;
1730 }
1731
1732 int consumer_open_channel_packets(struct consumer_socket *socket, uint64_t key)
1733 {
1734         int ret;
1735         lttcomm_consumer_msg msg = {
1736                 .cmd_type = LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS,
1737                 .u = {},
1738         };
1739         msg.u.open_channel_packets.key = key;
1740
1741         LTTNG_ASSERT(socket);
1742
1743         DBG("Consumer open channel packets: channel key = %" PRIu64, key);
1744
1745         health_code_update();
1746
1747         pthread_mutex_lock(socket->lock);
1748         ret = consumer_send_msg(socket, &msg);
1749         pthread_mutex_unlock(socket->lock);
1750         if (ret < 0) {
1751                 goto error_socket;
1752         }
1753
1754 error_socket:
1755         health_code_update();
1756         return ret;
1757 }
1758
1759 int consumer_clear_channel(struct consumer_socket *socket, uint64_t key)
1760 {
1761         int ret;
1762         struct lttcomm_consumer_msg msg;
1763
1764         LTTNG_ASSERT(socket);
1765
1766         DBG("Consumer clear channel %" PRIu64, key);
1767
1768         memset(&msg, 0, sizeof(msg));
1769         msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL;
1770         msg.u.clear_channel.key = key;
1771
1772         health_code_update();
1773
1774         pthread_mutex_lock(socket->lock);
1775         ret = consumer_send_msg(socket, &msg);
1776         if (ret < 0) {
1777                 goto error_socket;
1778         }
1779
1780 error_socket:
1781         pthread_mutex_unlock(socket->lock);
1782
1783         health_code_update();
1784         return ret;
1785 }
1786
1787 int consumer_init(struct consumer_socket *socket, const lttng_uuid& sessiond_uuid)
1788 {
1789         int ret;
1790         struct lttcomm_consumer_msg msg = {
1791                 .cmd_type = LTTNG_CONSUMER_INIT,
1792                 .u = {},
1793         };
1794
1795         LTTNG_ASSERT(socket);
1796
1797         DBG("Sending consumer initialization command");
1798         std::copy(sessiond_uuid.begin(), sessiond_uuid.end(), msg.u.init.sessiond_uuid);
1799
1800         health_code_update();
1801         ret = consumer_send_msg(socket, &msg);
1802         if (ret < 0) {
1803                 goto error;
1804         }
1805
1806 error:
1807         health_code_update();
1808         return ret;
1809 }
1810
1811 /*
1812  * Ask the consumer to create a new chunk for a given session.
1813  *
1814  * Called with the consumer socket lock held.
1815  */
1816 int consumer_create_trace_chunk(struct consumer_socket *socket,
1817                                 uint64_t relayd_id,
1818                                 uint64_t session_id,
1819                                 struct lttng_trace_chunk *chunk,
1820                                 const char *domain_subdir)
1821 {
1822         int ret;
1823         enum lttng_trace_chunk_status chunk_status;
1824         struct lttng_credentials chunk_credentials;
1825         const struct lttng_directory_handle *chunk_directory_handle = nullptr;
1826         struct lttng_directory_handle *domain_handle = nullptr;
1827         int domain_dirfd;
1828         const char *chunk_name;
1829         bool chunk_name_overridden;
1830         uint64_t chunk_id;
1831         time_t creation_timestamp;
1832         char creation_timestamp_buffer[ISO8601_STR_LEN];
1833         const char *creation_timestamp_str = "(none)";
1834         const bool chunk_has_local_output = relayd_id == -1ULL;
1835         enum lttng_trace_chunk_status tc_status;
1836         struct lttcomm_consumer_msg msg = {
1837                 .cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
1838                 .u = {},
1839         };
1840         msg.u.create_trace_chunk.session_id = session_id;
1841
1842         LTTNG_ASSERT(socket);
1843         LTTNG_ASSERT(chunk);
1844
1845         if (relayd_id != -1ULL) {
1846                 LTTNG_OPTIONAL_SET(&msg.u.create_trace_chunk.relayd_id, relayd_id);
1847         }
1848
1849         chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, &chunk_name_overridden);
1850         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1851             chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1852                 ERR("Failed to get name of trace chunk");
1853                 ret = -LTTNG_ERR_FATAL;
1854                 goto error;
1855         }
1856         if (chunk_name_overridden) {
1857                 ret = lttng_strncpy(msg.u.create_trace_chunk.override_name,
1858                                     chunk_name,
1859                                     sizeof(msg.u.create_trace_chunk.override_name));
1860                 if (ret) {
1861                         ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1862                             chunk_name);
1863                         ret = -LTTNG_ERR_FATAL;
1864                         goto error;
1865                 }
1866         }
1867
1868         chunk_status = lttng_trace_chunk_get_creation_timestamp(chunk, &creation_timestamp);
1869         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1870                 ret = -LTTNG_ERR_FATAL;
1871                 goto error;
1872         }
1873         msg.u.create_trace_chunk.creation_timestamp = (uint64_t) creation_timestamp;
1874         /* Only used for logging purposes. */
1875         ret = time_to_iso8601_str(
1876                 creation_timestamp, creation_timestamp_buffer, sizeof(creation_timestamp_buffer));
1877         creation_timestamp_str = !ret ? creation_timestamp_buffer : "(formatting error)";
1878
1879         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1880         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1881                 /*
1882                  * Anonymous trace chunks should never be transmitted
1883                  * to remote peers (consumerd and relayd). They are used
1884                  * internally for backward-compatibility purposes.
1885                  */
1886                 ret = -LTTNG_ERR_FATAL;
1887                 goto error;
1888         }
1889         msg.u.create_trace_chunk.chunk_id = chunk_id;
1890
1891         if (chunk_has_local_output) {
1892                 chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle(
1893                         chunk, &chunk_directory_handle);
1894                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1895                         ret = -LTTNG_ERR_FATAL;
1896                         goto error;
1897                 }
1898                 chunk_status = lttng_trace_chunk_get_credentials(chunk, &chunk_credentials);
1899                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1900                         /*
1901                          * Not associating credentials to a sessiond chunk is a
1902                          * fatal internal error.
1903                          */
1904                         ret = -LTTNG_ERR_FATAL;
1905                         goto error;
1906                 }
1907                 tc_status = lttng_trace_chunk_create_subdirectory(chunk, domain_subdir);
1908                 if (tc_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1909                         PERROR("Failed to create chunk domain output directory \"%s\"",
1910                                domain_subdir);
1911                         ret = -LTTNG_ERR_FATAL;
1912                         goto error;
1913                 }
1914                 domain_handle = lttng_directory_handle_create_from_handle(domain_subdir,
1915                                                                           chunk_directory_handle);
1916                 if (!domain_handle) {
1917                         ret = -LTTNG_ERR_FATAL;
1918                         goto error;
1919                 }
1920
1921                 /*
1922                  * This will only compile on platforms that support
1923                  * dirfd (POSIX.2008). This is fine as the session daemon
1924                  * is only built for such platforms.
1925                  *
1926                  * The ownership of the chunk directory handle's is maintained
1927                  * by the trace chunk.
1928                  */
1929                 domain_dirfd = lttng_directory_handle_get_dirfd(domain_handle);
1930                 LTTNG_ASSERT(domain_dirfd >= 0);
1931
1932                 msg.u.create_trace_chunk.credentials.value.uid =
1933                         lttng_credentials_get_uid(&chunk_credentials);
1934                 msg.u.create_trace_chunk.credentials.value.gid =
1935                         lttng_credentials_get_gid(&chunk_credentials);
1936                 msg.u.create_trace_chunk.credentials.is_set = 1;
1937         }
1938
1939         DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1940             ", session_id = %" PRIu64 ", chunk_id = %" PRIu64 ", creation_timestamp = %s",
1941             relayd_id,
1942             session_id,
1943             chunk_id,
1944             creation_timestamp_str);
1945         health_code_update();
1946         ret = consumer_send_msg(socket, &msg);
1947         health_code_update();
1948         if (ret < 0) {
1949                 ERR("Trace chunk creation error on consumer");
1950                 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1951                 goto error;
1952         }
1953
1954         if (chunk_has_local_output) {
1955                 DBG("Sending trace chunk domain directory fd to consumer");
1956                 health_code_update();
1957                 ret = consumer_send_fds(socket, &domain_dirfd, 1);
1958                 health_code_update();
1959                 if (ret < 0) {
1960                         ERR("Trace chunk creation error on consumer");
1961                         ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1962                         goto error;
1963                 }
1964         }
1965 error:
1966         lttng_directory_handle_put(domain_handle);
1967         return ret;
1968 }
1969
1970 /*
1971  * Ask the consumer to close a trace chunk for a given session.
1972  *
1973  * Called with the consumer socket lock held.
1974  */
1975 int consumer_close_trace_chunk(struct consumer_socket *socket,
1976                                uint64_t relayd_id,
1977                                uint64_t session_id,
1978                                struct lttng_trace_chunk *chunk,
1979                                char *closed_trace_chunk_path)
1980 {
1981         int ret;
1982         enum lttng_trace_chunk_status chunk_status;
1983         lttcomm_consumer_msg msg = {
1984                 .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
1985                 .u = {},
1986         };
1987         msg.u.close_trace_chunk.session_id = session_id;
1988
1989         struct lttcomm_consumer_close_trace_chunk_reply reply;
1990         uint64_t chunk_id;
1991         time_t close_timestamp;
1992         enum lttng_trace_chunk_command_type close_command;
1993         const char *close_command_name = "none";
1994         struct lttng_dynamic_buffer path_reception_buffer;
1995
1996         LTTNG_ASSERT(socket);
1997         lttng_dynamic_buffer_init(&path_reception_buffer);
1998
1999         if (relayd_id != -1ULL) {
2000                 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.relayd_id, relayd_id);
2001         }
2002
2003         chunk_status = lttng_trace_chunk_get_close_command(chunk, &close_command);
2004         switch (chunk_status) {
2005         case LTTNG_TRACE_CHUNK_STATUS_OK:
2006                 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command,
2007                                    (uint32_t) close_command);
2008                 break;
2009         case LTTNG_TRACE_CHUNK_STATUS_NONE:
2010                 break;
2011         default:
2012                 ERR("Failed to get trace chunk close command");
2013                 ret = -1;
2014                 goto error;
2015         }
2016
2017         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2018         /*
2019          * Anonymous trace chunks should never be transmitted to remote peers
2020          * (consumerd and relayd). They are used internally for
2021          * backward-compatibility purposes.
2022          */
2023         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2024         msg.u.close_trace_chunk.chunk_id = chunk_id;
2025
2026         chunk_status = lttng_trace_chunk_get_close_timestamp(chunk, &close_timestamp);
2027         /*
2028          * A trace chunk should be closed locally before being closed remotely.
2029          * Otherwise, the close timestamp would never be transmitted to the
2030          * peers.
2031          */
2032         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2033         msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp;
2034
2035         if (msg.u.close_trace_chunk.close_command.is_set) {
2036                 close_command_name = lttng_trace_chunk_command_type_get_name(close_command);
2037         }
2038         DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
2039             ", session_id = %" PRIu64 ", chunk_id = %" PRIu64 ", close command = \"%s\"",
2040             relayd_id,
2041             session_id,
2042             chunk_id,
2043             close_command_name);
2044
2045         health_code_update();
2046         ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg));
2047         if (ret < 0) {
2048                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2049                 goto error;
2050         }
2051         ret = consumer_socket_recv(socket, &reply, sizeof(reply));
2052         if (ret < 0) {
2053                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2054                 goto error;
2055         }
2056         if (reply.path_length >= LTTNG_PATH_MAX) {
2057                 ERR("Invalid path returned by relay daemon: %" PRIu32
2058                     "bytes exceeds maximal allowed length of %d bytes",
2059                     reply.path_length,
2060                     LTTNG_PATH_MAX);
2061                 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2062                 goto error;
2063         }
2064         ret = lttng_dynamic_buffer_set_size(&path_reception_buffer, reply.path_length);
2065         if (ret) {
2066                 ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command");
2067                 ret = -LTTNG_ERR_NOMEM;
2068                 goto error;
2069         }
2070         ret = consumer_socket_recv(socket, path_reception_buffer.data, path_reception_buffer.size);
2071         if (ret < 0) {
2072                 ERR("Communication error while receiving path of closed trace chunk");
2073                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2074                 goto error;
2075         }
2076         if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') {
2077                 ERR("Invalid path returned by relay daemon: not null-terminated");
2078                 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2079                 goto error;
2080         }
2081         if (closed_trace_chunk_path) {
2082                 /*
2083                  * closed_trace_chunk_path is assumed to have a length >=
2084                  * LTTNG_PATH_MAX
2085                  */
2086                 memcpy(closed_trace_chunk_path,
2087                        path_reception_buffer.data,
2088                        path_reception_buffer.size);
2089         }
2090 error:
2091         lttng_dynamic_buffer_reset(&path_reception_buffer);
2092         health_code_update();
2093         return ret;
2094 }
2095
2096 /*
2097  * Ask the consumer if a trace chunk exists.
2098  *
2099  * Called with the consumer socket lock held.
2100  * Returns 0 on success, or a negative value on error.
2101  */
2102 int consumer_trace_chunk_exists(struct consumer_socket *socket,
2103                                 uint64_t relayd_id,
2104                                 uint64_t session_id,
2105                                 struct lttng_trace_chunk *chunk,
2106                                 enum consumer_trace_chunk_exists_status *result)
2107 {
2108         int ret;
2109         enum lttng_trace_chunk_status chunk_status;
2110         lttcomm_consumer_msg msg = {
2111                 .cmd_type = LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,
2112                 .u = {},
2113         };
2114         msg.u.trace_chunk_exists.session_id = session_id;
2115
2116         uint64_t chunk_id;
2117         const char *consumer_reply_str;
2118
2119         LTTNG_ASSERT(socket);
2120
2121         if (relayd_id != -1ULL) {
2122                 LTTNG_OPTIONAL_SET(&msg.u.trace_chunk_exists.relayd_id, relayd_id);
2123         }
2124
2125         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2126         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2127                 /*
2128                  * Anonymous trace chunks should never be transmitted
2129                  * to remote peers (consumerd and relayd). They are used
2130                  * internally for backward-compatibility purposes.
2131                  */
2132                 ret = -LTTNG_ERR_FATAL;
2133                 goto error;
2134         }
2135         msg.u.trace_chunk_exists.chunk_id = chunk_id;
2136
2137         DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
2138             ", session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2139             relayd_id,
2140             session_id,
2141             chunk_id);
2142
2143         health_code_update();
2144         ret = consumer_send_msg(socket, &msg);
2145         switch (-ret) {
2146         case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK:
2147                 consumer_reply_str = "unknown trace chunk";
2148                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK;
2149                 break;
2150         case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL:
2151                 consumer_reply_str = "trace chunk exists locally";
2152                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL;
2153                 break;
2154         case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE:
2155                 consumer_reply_str = "trace chunk exists on remote peer";
2156                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE;
2157                 break;
2158         default:
2159                 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
2160                 ret = -1;
2161                 goto error;
2162         }
2163         DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s", consumer_reply_str);
2164         ret = 0;
2165 error:
2166         health_code_update();
2167         return ret;
2168 }