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 <stdio.h>
  11 #include <stdlib.h>
  12 #include <string.h>
  13 #include <sys/stat.h>
  14 #include <sys/types.h>
  15 #include <unistd.h>
  16 #include <inttypes.h>
  17
  18 #include <common/common.h>
  19 #include <common/defaults.h>
  20 #include <common/uri.h>
  21 #include <common/relayd/relayd.h>
  22 #include <common/string-utils/format.h>
  23
  24 #include "consumer.h"
  25 #include "health-sessiond.h"
  26 #include "ust-app.h"
  27 #include "utils.h"
  28 #include "lttng-sessiond.h"
  29
  30 /*
  31  * Return allocated full pathname of the session using the consumer trace path
  32  * and subdir if available.
  33  *
  34  * The caller can safely free(3) the returned value. On error, NULL is
  35  * returned.
  36  */
  37 char *setup_channel_trace_path(struct consumer_output *consumer,
  38                 const char *session_path, size_t *consumer_path_offset)
  39 {
  40         int ret;
  41         char *pathname;
  42
  43         LTTNG_ASSERT(consumer);
  44         LTTNG_ASSERT(session_path);
  45
  46         health_code_update();
  47
  48         /*
  49          * Allocate the string ourself to make sure we never exceed
  50          * LTTNG_PATH_MAX.
  51          */
  52         pathname = zmalloc(LTTNG_PATH_MAX);
  53         if (!pathname) {
  54                 goto error;
  55         }
  56
  57         /* Get correct path name destination */
  58         if (consumer->type == CONSUMER_DST_NET &&
  59                         consumer->relay_major_version == 2 &&
  60                         consumer->relay_minor_version < 11) {
  61                 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s/%s/%s",
  62                                 consumer->dst.net.base_dir,
  63                                 consumer->chunk_path, consumer->domain_subdir,
  64                                 session_path);
  65                 *consumer_path_offset = 0;
  66         } else {
  67                 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s/%s",
  68                                 consumer->domain_subdir, session_path);
  69                 *consumer_path_offset = strlen(consumer->domain_subdir) + 1;
  70         }
  71         DBG3("Consumer trace path relative to current trace chunk: \"%s\"",
  72                         pathname);
  73         if (ret < 0) {
  74                 PERROR("Failed to format channel path");
  75                 goto error;
  76         } else if (ret >= LTTNG_PATH_MAX) {
  77                 ERR("Truncation occurred while formatting channel path");
  78                 goto error;
  79         }
  80
  81         return pathname;
  82 error:
  83         free(pathname);
  84         return NULL;
  85 }
  86
  87 /*
  88  * Send a data payload using a given consumer socket of size len.
  89  *
  90  * The consumer socket lock MUST be acquired before calling this since this
  91  * function can change the fd value.
  92  *
  93  * Return 0 on success else a negative value on error.
  94  */
  95 int consumer_socket_send(
  96                 struct consumer_socket *socket, const void *msg, size_t len)
  97 {
  98         int fd;
  99         ssize_t size;
 100
 101         LTTNG_ASSERT(socket);
 102         LTTNG_ASSERT(socket->fd_ptr);
 103         LTTNG_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         LTTNG_ASSERT(socket);
 146         LTTNG_ASSERT(socket->fd_ptr);
 147         LTTNG_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         LTTNG_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         LTTNG_ASSERT(sock);
 222         LTTNG_ASSERT(stream_count);
 223         LTTNG_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         LTTNG_ASSERT(consumer);
 256         LTTNG_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         LTTNG_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         LTTNG_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(&the_ust_consumerd64_fd);
 373                 break;
 374         case 32:
 375                 consumer_fd = uatomic_read(&the_ust_consumerd32_fd);
 376                 break;
 377         default:
 378                 abort();
 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         LTTNG_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         LTTNG_ASSERT(sock);
 449         LTTNG_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         LTTNG_ASSERT(sock);
 465         LTTNG_ASSERT(consumer);
 466
 467         iter.iter.node = &sock->node.node;
 468         ret = lttng_ht_del(consumer->socks, &iter);
 469         LTTNG_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         LTTNG_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         LTTNG_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         LTTNG_ASSERT(dst);
 646         LTTNG_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         LTTNG_ASSERT(output);
 694         LTTNG_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         LTTNG_ASSERT(fds);
 842         LTTNG_ASSERT(sock);
 843         LTTNG_ASSERT(nb_fd > 0);
 844         LTTNG_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                 const struct lttcomm_consumer_msg *msg)
 865 {
 866         int ret;
 867
 868         LTTNG_ASSERT(msg);
 869         LTTNG_ASSERT(sock);
 870         LTTNG_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         LTTNG_ASSERT(msg);
 894         LTTNG_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         LTTNG_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                 LTTNG_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 =
 954                         lttng_credentials_get_uid(buffer_credentials);
 955         msg->u.ask_channel.buffer_credentials.gid =
 956                         lttng_credentials_get_gid(buffer_credentials);
 957
 958         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 959         msg->u.ask_channel.subbuf_size = subbuf_size;
 960         msg->u.ask_channel.num_subbuf = num_subbuf ;
 961         msg->u.ask_channel.overwrite = overwrite;
 962         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 963         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 964         msg->u.ask_channel.live_timer_interval = live_timer_interval;
 965         msg->u.ask_channel.is_live = is_in_live_session;
 966         msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
 967         msg->u.ask_channel.output = output;
 968         msg->u.ask_channel.type = type;
 969         msg->u.ask_channel.session_id = session_id;
 970         msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
 971         msg->u.ask_channel.relayd_id = relayd_id;
 972         msg->u.ask_channel.key = key;
 973         msg->u.ask_channel.chan_id = chan_id;
 974         msg->u.ask_channel.tracefile_size = tracefile_size;
 975         msg->u.ask_channel.tracefile_count = tracefile_count;
 976         msg->u.ask_channel.monitor = monitor;
 977         msg->u.ask_channel.ust_app_uid = ust_app_uid;
 978         msg->u.ask_channel.blocking_timeout = blocking_timeout;
 979
 980         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 981
 982         if (pathname) {
 983                 strncpy(msg->u.ask_channel.pathname, pathname,
 984                                 sizeof(msg->u.ask_channel.pathname));
 985                 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 986         }
 987
 988         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 989         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 990
 991         if (root_shm_path) {
 992                 strncpy(msg->u.ask_channel.root_shm_path, root_shm_path,
 993                         sizeof(msg->u.ask_channel.root_shm_path));
 994                 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] = '\0';
 995         }
 996         if (shm_path) {
 997                 strncpy(msg->u.ask_channel.shm_path, shm_path,
 998                         sizeof(msg->u.ask_channel.shm_path));
 999                 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
1000         }
1001 }
1002
1003 /*
1004  * Init channel communication message structure.
1005  */
1006 void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg,
1007                 uint64_t channel_key,
1008                 uint64_t session_id,
1009                 const char *pathname,
1010                 uid_t uid,
1011                 gid_t gid,
1012                 uint64_t relayd_id,
1013                 const char *name,
1014                 unsigned int nb_init_streams,
1015                 enum lttng_event_output output,
1016                 int type,
1017                 uint64_t tracefile_size,
1018                 uint64_t tracefile_count,
1019                 unsigned int monitor,
1020                 unsigned int live_timer_interval,
1021                 bool is_in_live_session,
1022                 unsigned int monitor_timer_interval,
1023                 struct lttng_trace_chunk *trace_chunk)
1024 {
1025         LTTNG_ASSERT(msg);
1026
1027         /* Zeroed structure */
1028         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1029
1030         if (trace_chunk) {
1031                 uint64_t chunk_id;
1032                 enum lttng_trace_chunk_status chunk_status;
1033
1034                 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
1035                 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
1036                 LTTNG_OPTIONAL_SET(&msg->u.channel.chunk_id, chunk_id);
1037         }
1038
1039         /* Send channel */
1040         msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
1041         msg->u.channel.channel_key = channel_key;
1042         msg->u.channel.session_id = session_id;
1043         msg->u.channel.relayd_id = relayd_id;
1044         msg->u.channel.nb_init_streams = nb_init_streams;
1045         msg->u.channel.output = output;
1046         msg->u.channel.type = type;
1047         msg->u.channel.tracefile_size = tracefile_size;
1048         msg->u.channel.tracefile_count = tracefile_count;
1049         msg->u.channel.monitor = monitor;
1050         msg->u.channel.live_timer_interval = live_timer_interval;
1051         msg->u.channel.is_live = is_in_live_session;
1052         msg->u.channel.monitor_timer_interval = monitor_timer_interval;
1053
1054         strncpy(msg->u.channel.pathname, pathname,
1055                         sizeof(msg->u.channel.pathname));
1056         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
1057
1058         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
1059         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
1060 }
1061
1062 /*
1063  * Init stream communication message structure.
1064  */
1065 void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg,
1066                 uint64_t channel_key,
1067                 uint64_t stream_key,
1068                 int32_t cpu)
1069 {
1070         LTTNG_ASSERT(msg);
1071
1072         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1073
1074         msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM;
1075         msg->u.stream.channel_key = channel_key;
1076         msg->u.stream.stream_key = stream_key;
1077         msg->u.stream.cpu = cpu;
1078 }
1079
1080 void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
1081                 enum lttng_consumer_command cmd,
1082                 uint64_t channel_key, uint64_t net_seq_idx)
1083 {
1084         LTTNG_ASSERT(msg);
1085
1086         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1087
1088         msg->cmd_type = cmd;
1089         msg->u.sent_streams.channel_key = channel_key;
1090         msg->u.sent_streams.net_seq_idx = net_seq_idx;
1091 }
1092
1093 /*
1094  * Send stream communication structure to the consumer.
1095  */
1096 int consumer_send_stream(struct consumer_socket *sock,
1097                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
1098                 const int *fds, size_t nb_fd)
1099 {
1100         int ret;
1101
1102         LTTNG_ASSERT(msg);
1103         LTTNG_ASSERT(dst);
1104         LTTNG_ASSERT(sock);
1105         LTTNG_ASSERT(fds);
1106
1107         ret = consumer_send_msg(sock, msg);
1108         if (ret < 0) {
1109                 goto error;
1110         }
1111
1112         ret = consumer_send_fds(sock, fds, nb_fd);
1113         if (ret < 0) {
1114                 goto error;
1115         }
1116
1117 error:
1118         return ret;
1119 }
1120
1121 /*
1122  * Send relayd socket to consumer associated with a session name.
1123  *
1124  * The consumer socket lock must be held by the caller.
1125  *
1126  * On success return positive value. On error, negative value.
1127  */
1128 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
1129                 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
1130                 enum lttng_stream_type type, uint64_t session_id,
1131                 const char *session_name, const char *hostname,
1132                 const char *base_path, int session_live_timer,
1133                 const uint64_t *current_chunk_id, time_t session_creation_time,
1134                 bool session_name_contains_creation_time)
1135 {
1136         int ret;
1137         struct lttcomm_consumer_msg msg;
1138
1139         /* Code flow error. Safety net. */
1140         LTTNG_ASSERT(rsock);
1141         LTTNG_ASSERT(consumer);
1142         LTTNG_ASSERT(consumer_sock);
1143
1144         memset(&msg, 0, sizeof(msg));
1145         /* Bail out if consumer is disabled */
1146         if (!consumer->enabled) {
1147                 ret = LTTNG_OK;
1148                 goto error;
1149         }
1150
1151         if (type == LTTNG_STREAM_CONTROL) {
1152                 char output_path[LTTNG_PATH_MAX] = {};
1153                 uint64_t relayd_session_id;
1154
1155                 ret = relayd_create_session(rsock, &relayd_session_id,
1156                                 session_name, hostname, base_path,
1157                                 session_live_timer, consumer->snapshot,
1158                                 session_id, the_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         LTTNG_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         LTTNG_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         LTTNG_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         LTTNG_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         LTTNG_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         LTTNG_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         LTTNG_ASSERT(socket);
1511         LTTNG_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         LTTNG_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         LTTNG_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         LTTNG_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_open_channel_packets(struct consumer_socket *socket, uint64_t key)
1725 {
1726         int ret;
1727         const struct lttcomm_consumer_msg msg = {
1728                 .cmd_type = LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS,
1729                 .u.open_channel_packets.key = key,
1730         };
1731
1732         LTTNG_ASSERT(socket);
1733
1734         DBG("Consumer open channel packets: channel key = %" PRIu64, key);
1735
1736         health_code_update();
1737
1738         pthread_mutex_lock(socket->lock);
1739         ret = consumer_send_msg(socket, &msg);
1740         pthread_mutex_unlock(socket->lock);
1741         if (ret < 0) {
1742                 goto error_socket;
1743         }
1744
1745 error_socket:
1746         health_code_update();
1747         return ret;
1748 }
1749
1750 int consumer_clear_channel(struct consumer_socket *socket, uint64_t key)
1751 {
1752         int ret;
1753         struct lttcomm_consumer_msg msg;
1754
1755         LTTNG_ASSERT(socket);
1756
1757         DBG("Consumer clear channel %" PRIu64, key);
1758
1759         memset(&msg, 0, sizeof(msg));
1760         msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL;
1761         msg.u.clear_channel.key = key;
1762
1763         health_code_update();
1764
1765         pthread_mutex_lock(socket->lock);
1766         ret = consumer_send_msg(socket, &msg);
1767         if (ret < 0) {
1768                 goto error_socket;
1769         }
1770
1771 error_socket:
1772         pthread_mutex_unlock(socket->lock);
1773
1774         health_code_update();
1775         return ret;
1776 }
1777
1778 int consumer_init(struct consumer_socket *socket,
1779                 const lttng_uuid sessiond_uuid)
1780 {
1781         int ret;
1782         struct lttcomm_consumer_msg msg = {
1783                 .cmd_type = LTTNG_CONSUMER_INIT,
1784         };
1785
1786         LTTNG_ASSERT(socket);
1787
1788         DBG("Sending consumer initialization command");
1789         lttng_uuid_copy(msg.u.init.sessiond_uuid, sessiond_uuid);
1790
1791         health_code_update();
1792         ret = consumer_send_msg(socket, &msg);
1793         if (ret < 0) {
1794                 goto error;
1795         }
1796
1797 error:
1798         health_code_update();
1799         return ret;
1800 }
1801
1802 /*
1803  * Ask the consumer to create a new chunk for a given session.
1804  *
1805  * Called with the consumer socket lock held.
1806  */
1807 int consumer_create_trace_chunk(struct consumer_socket *socket,
1808                 uint64_t relayd_id, uint64_t session_id,
1809                 struct lttng_trace_chunk *chunk,
1810                 const char *domain_subdir)
1811 {
1812         int ret;
1813         enum lttng_trace_chunk_status chunk_status;
1814         struct lttng_credentials chunk_credentials;
1815         const struct lttng_directory_handle *chunk_directory_handle = NULL;
1816         struct lttng_directory_handle *domain_handle = NULL;
1817         int domain_dirfd;
1818         const char *chunk_name;
1819         bool chunk_name_overridden;
1820         uint64_t chunk_id;
1821         time_t creation_timestamp;
1822         char creation_timestamp_buffer[ISO8601_STR_LEN];
1823         const char *creation_timestamp_str = "(none)";
1824         const bool chunk_has_local_output = relayd_id == -1ULL;
1825         enum lttng_trace_chunk_status tc_status;
1826         struct lttcomm_consumer_msg msg = {
1827                 .cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
1828                 .u.create_trace_chunk.session_id = session_id,
1829         };
1830
1831         LTTNG_ASSERT(socket);
1832         LTTNG_ASSERT(chunk);
1833
1834         if (relayd_id != -1ULL) {
1835                 LTTNG_OPTIONAL_SET(&msg.u.create_trace_chunk.relayd_id,
1836                                 relayd_id);
1837         }
1838
1839         chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name,
1840                         &chunk_name_overridden);
1841         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1842                         chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1843                 ERR("Failed to get name of trace chunk");
1844                 ret = -LTTNG_ERR_FATAL;
1845                 goto error;
1846         }
1847         if (chunk_name_overridden) {
1848                 ret = lttng_strncpy(msg.u.create_trace_chunk.override_name,
1849                                 chunk_name,
1850                                 sizeof(msg.u.create_trace_chunk.override_name));
1851                 if (ret) {
1852                         ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1853                                         chunk_name);
1854                         ret = -LTTNG_ERR_FATAL;
1855                         goto error;
1856                 }
1857         }
1858
1859         chunk_status = lttng_trace_chunk_get_creation_timestamp(chunk,
1860                         &creation_timestamp);
1861         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1862                 ret = -LTTNG_ERR_FATAL;
1863                 goto error;
1864         }
1865         msg.u.create_trace_chunk.creation_timestamp =
1866                         (uint64_t) creation_timestamp;
1867         /* Only used for logging purposes. */
1868         ret = time_to_iso8601_str(creation_timestamp,
1869                         creation_timestamp_buffer,
1870                         sizeof(creation_timestamp_buffer));
1871         creation_timestamp_str = !ret ? creation_timestamp_buffer :
1872                         "(formatting error)";
1873
1874         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1875         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1876                 /*
1877                  * Anonymous trace chunks should never be transmitted
1878                  * to remote peers (consumerd and relayd). They are used
1879                  * internally for backward-compatibility purposes.
1880                  */
1881                 ret = -LTTNG_ERR_FATAL;
1882                 goto error;
1883         }
1884         msg.u.create_trace_chunk.chunk_id = chunk_id;
1885
1886         if (chunk_has_local_output) {
1887                 chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle(
1888                                 chunk, &chunk_directory_handle);
1889                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1890                         ret = -LTTNG_ERR_FATAL;
1891                         goto error;
1892                 }
1893                 chunk_status = lttng_trace_chunk_get_credentials(
1894                                 chunk, &chunk_credentials);
1895                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1896                         /*
1897                          * Not associating credentials to a sessiond chunk is a
1898                          * fatal internal error.
1899                          */
1900                         ret = -LTTNG_ERR_FATAL;
1901                         goto error;
1902                 }
1903                 tc_status = lttng_trace_chunk_create_subdirectory(
1904                                 chunk, domain_subdir);
1905                 if (tc_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1906                         PERROR("Failed to create chunk domain output directory \"%s\"",
1907                                 domain_subdir);
1908                         ret = -LTTNG_ERR_FATAL;
1909                         goto error;
1910                 }
1911                 domain_handle = lttng_directory_handle_create_from_handle(
1912                                 domain_subdir,
1913                                 chunk_directory_handle);
1914                 if (!domain_handle) {
1915                         ret = -LTTNG_ERR_FATAL;
1916                         goto error;
1917                 }
1918
1919                 /*
1920                  * This will only compile on platforms that support
1921                  * dirfd (POSIX.2008). This is fine as the session daemon
1922                  * is only built for such platforms.
1923                  *
1924                  * The ownership of the chunk directory handle's is maintained
1925                  * by the trace chunk.
1926                  */
1927                 domain_dirfd = lttng_directory_handle_get_dirfd(
1928                                 domain_handle);
1929                 LTTNG_ASSERT(domain_dirfd >= 0);
1930
1931                 msg.u.create_trace_chunk.credentials.value.uid =
1932                                 lttng_credentials_get_uid(&chunk_credentials);
1933                 msg.u.create_trace_chunk.credentials.value.gid =
1934                                 lttng_credentials_get_gid(&chunk_credentials);
1935                 msg.u.create_trace_chunk.credentials.is_set = 1;
1936         }
1937
1938         DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1939                         ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
1940                         ", creation_timestamp = %s",
1941                         relayd_id, session_id, chunk_id,
1942                         creation_timestamp_str);
1943         health_code_update();
1944         ret = consumer_send_msg(socket, &msg);
1945         health_code_update();
1946         if (ret < 0) {
1947                 ERR("Trace chunk creation error on consumer");
1948                 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1949                 goto error;
1950         }
1951
1952         if (chunk_has_local_output) {
1953                 DBG("Sending trace chunk domain directory fd to consumer");
1954                 health_code_update();
1955                 ret = consumer_send_fds(socket, &domain_dirfd, 1);
1956                 health_code_update();
1957                 if (ret < 0) {
1958                         ERR("Trace chunk creation error on consumer");
1959                         ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1960                         goto error;
1961                 }
1962         }
1963 error:
1964         lttng_directory_handle_put(domain_handle);
1965         return ret;
1966 }
1967
1968 /*
1969  * Ask the consumer to close a trace chunk for a given session.
1970  *
1971  * Called with the consumer socket lock held.
1972  */
1973 int consumer_close_trace_chunk(struct consumer_socket *socket,
1974                 uint64_t relayd_id, uint64_t session_id,
1975                 struct lttng_trace_chunk *chunk,
1976                 char *closed_trace_chunk_path)
1977 {
1978         int ret;
1979         enum lttng_trace_chunk_status chunk_status;
1980         struct lttcomm_consumer_msg msg = {
1981                         .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
1982                         .u.close_trace_chunk.session_id = session_id,
1983         };
1984         struct lttcomm_consumer_close_trace_chunk_reply reply;
1985         uint64_t chunk_id;
1986         time_t close_timestamp;
1987         enum lttng_trace_chunk_command_type close_command;
1988         const char *close_command_name = "none";
1989         struct lttng_dynamic_buffer path_reception_buffer;
1990
1991         LTTNG_ASSERT(socket);
1992         lttng_dynamic_buffer_init(&path_reception_buffer);
1993
1994         if (relayd_id != -1ULL) {
1995                 LTTNG_OPTIONAL_SET(
1996                                 &msg.u.close_trace_chunk.relayd_id, relayd_id);
1997         }
1998
1999         chunk_status = lttng_trace_chunk_get_close_command(
2000                         chunk, &close_command);
2001         switch (chunk_status) {
2002         case LTTNG_TRACE_CHUNK_STATUS_OK:
2003                 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command,
2004                                 (uint32_t) close_command);
2005                 break;
2006         case LTTNG_TRACE_CHUNK_STATUS_NONE:
2007                 break;
2008         default:
2009                 ERR("Failed to get trace chunk close command");
2010                 ret = -1;
2011                 goto error;
2012         }
2013
2014         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2015         /*
2016          * Anonymous trace chunks should never be transmitted to remote peers
2017          * (consumerd and relayd). They are used internally for
2018          * backward-compatibility purposes.
2019          */
2020         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2021         msg.u.close_trace_chunk.chunk_id = chunk_id;
2022
2023         chunk_status = lttng_trace_chunk_get_close_timestamp(chunk,
2024                         &close_timestamp);
2025         /*
2026          * A trace chunk should be closed locally before being closed remotely.
2027          * Otherwise, the close timestamp would never be transmitted to the
2028          * peers.
2029          */
2030         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2031         msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp;
2032
2033         if (msg.u.close_trace_chunk.close_command.is_set) {
2034                 close_command_name = lttng_trace_chunk_command_type_get_name(
2035                                 close_command);
2036         }
2037         DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
2038                         ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
2039                         ", close command = \"%s\"",
2040                         relayd_id, session_id, chunk_id, close_command_name);
2041
2042         health_code_update();
2043         ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg));
2044         if (ret < 0) {
2045                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2046                 goto error;
2047         }
2048         ret = consumer_socket_recv(socket, &reply, sizeof(reply));
2049         if (ret < 0) {
2050                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2051                 goto error;
2052         }
2053         if (reply.path_length >= LTTNG_PATH_MAX) {
2054                 ERR("Invalid path returned by relay daemon: %" PRIu32 "bytes exceeds maximal allowed length of %d bytes",
2055                                 reply.path_length, LTTNG_PATH_MAX);
2056                 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2057                 goto error;
2058         }
2059         ret = lttng_dynamic_buffer_set_size(&path_reception_buffer,
2060                         reply.path_length);
2061         if (ret) {
2062                 ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command");
2063                 ret = -LTTNG_ERR_NOMEM;
2064                 goto error;
2065         }
2066         ret = consumer_socket_recv(socket, path_reception_buffer.data,
2067                         path_reception_buffer.size);
2068         if (ret < 0) {
2069                 ERR("Communication error while receiving path of closed trace chunk");
2070                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2071                 goto error;
2072         }
2073         if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') {
2074                 ERR("Invalid path returned by relay daemon: not null-terminated");
2075                 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2076                 goto error;
2077         }
2078         if (closed_trace_chunk_path) {
2079                 /*
2080                  * closed_trace_chunk_path is assumed to have a length >=
2081                  * LTTNG_PATH_MAX
2082                  */
2083                 memcpy(closed_trace_chunk_path, path_reception_buffer.data,
2084                                 path_reception_buffer.size);
2085         }
2086 error:
2087         lttng_dynamic_buffer_reset(&path_reception_buffer);
2088         health_code_update();
2089         return ret;
2090 }
2091
2092 /*
2093  * Ask the consumer if a trace chunk exists.
2094  *
2095  * Called with the consumer socket lock held.
2096  * Returns 0 on success, or a negative value on error.
2097  */
2098 int consumer_trace_chunk_exists(struct consumer_socket *socket,
2099                 uint64_t relayd_id, uint64_t session_id,
2100                 struct lttng_trace_chunk *chunk,
2101                 enum consumer_trace_chunk_exists_status *result)
2102 {
2103         int ret;
2104         enum lttng_trace_chunk_status chunk_status;
2105         struct lttcomm_consumer_msg msg = {
2106                 .cmd_type = LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,
2107                 .u.trace_chunk_exists.session_id = session_id,
2108         };
2109         uint64_t chunk_id;
2110         const char *consumer_reply_str;
2111
2112         LTTNG_ASSERT(socket);
2113
2114         if (relayd_id != -1ULL) {
2115                 LTTNG_OPTIONAL_SET(&msg.u.trace_chunk_exists.relayd_id,
2116                                 relayd_id);
2117         }
2118
2119         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2120         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2121                 /*
2122                  * Anonymous trace chunks should never be transmitted
2123                  * to remote peers (consumerd and relayd). They are used
2124                  * internally for backward-compatibility purposes.
2125                  */
2126                 ret = -LTTNG_ERR_FATAL;
2127                 goto error;
2128         }
2129         msg.u.trace_chunk_exists.chunk_id = chunk_id;
2130
2131         DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
2132                         ", session_id = %" PRIu64
2133                         ", chunk_id = %" PRIu64, relayd_id, session_id, chunk_id);
2134
2135         health_code_update();
2136         ret = consumer_send_msg(socket, &msg);
2137         switch (-ret) {
2138         case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK:
2139                 consumer_reply_str = "unknown trace chunk";
2140                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK;
2141                 break;
2142         case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL:
2143                 consumer_reply_str = "trace chunk exists locally";
2144                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL;
2145                 break;
2146         case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE:
2147                 consumer_reply_str = "trace chunk exists on remote peer";
2148                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE;
2149                 break;
2150         default:
2151                 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
2152                 ret = -1;
2153                 goto error;
2154         }
2155         DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
2156                         consumer_reply_str);
2157         ret = 0;
2158 error:
2159         health_code_update();
2160         return ret;
2161 }