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