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