src/bin/lttng-sessiond/consumer.c

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