src/bin/lttng-sessiond/consumer.cpp

   1 /*
   2  * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
   3  * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * SPDX-License-Identifier: GPL-2.0-only
   6  *
   7  */
   8
   9 #define _LGPL_SOURCE
  10 #include <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 = (char *) 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 = (consumer_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 = (consumer_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         int fd;
1138         struct lttcomm_consumer_msg msg;
1139
1140         /* Code flow error. Safety net. */
1141         LTTNG_ASSERT(rsock);
1142         LTTNG_ASSERT(consumer);
1143         LTTNG_ASSERT(consumer_sock);
1144
1145         memset(&msg, 0, sizeof(msg));
1146         /* Bail out if consumer is disabled */
1147         if (!consumer->enabled) {
1148                 ret = LTTNG_OK;
1149                 goto error;
1150         }
1151
1152         if (type == LTTNG_STREAM_CONTROL) {
1153                 char output_path[LTTNG_PATH_MAX] = {};
1154                 uint64_t relayd_session_id;
1155
1156                 ret = relayd_create_session(rsock, &relayd_session_id,
1157                                 session_name, hostname, base_path,
1158                                 session_live_timer, consumer->snapshot,
1159                                 session_id, the_sessiond_uuid, current_chunk_id,
1160                                 session_creation_time,
1161                                 session_name_contains_creation_time,
1162                                 output_path);
1163                 if (ret < 0) {
1164                         /* Close the control socket. */
1165                         (void) relayd_close(rsock);
1166                         goto error;
1167                 }
1168                 msg.u.relayd_sock.relayd_session_id = relayd_session_id;
1169                 DBG("Created session on relay, output path reply: %s",
1170                         output_path);
1171         }
1172
1173         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1174         /*
1175          * Assign network consumer output index using the temporary consumer since
1176          * this call should only be made from within a set_consumer_uri() function
1177          * call in the session daemon.
1178          */
1179         msg.u.relayd_sock.net_index = consumer->net_seq_index;
1180         msg.u.relayd_sock.type = type;
1181         msg.u.relayd_sock.session_id = session_id;
1182         memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
1183
1184         DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
1185         ret = consumer_send_msg(consumer_sock, &msg);
1186         if (ret < 0) {
1187                 goto error;
1188         }
1189
1190         DBG3("Sending relayd socket file descriptor to consumer");
1191         fd = rsock->sock.fd;
1192         ret = consumer_send_fds(consumer_sock, &fd, 1);
1193         if (ret < 0) {
1194                 goto error;
1195         }
1196
1197         DBG2("Consumer relayd socket sent");
1198
1199 error:
1200         return ret;
1201 }
1202
1203 static
1204 int consumer_send_pipe(struct consumer_socket *consumer_sock,
1205                 enum lttng_consumer_command cmd, int pipe)
1206 {
1207         int ret;
1208         struct lttcomm_consumer_msg msg;
1209         const char *pipe_name;
1210         const char *command_name;
1211
1212         switch (cmd) {
1213         case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1214                 pipe_name = "channel monitor";
1215                 command_name = "SET_CHANNEL_MONITOR_PIPE";
1216                 break;
1217         default:
1218                 ERR("Unexpected command received in %s (cmd = %d)", __func__,
1219                                 (int) cmd);
1220                 abort();
1221         }
1222
1223         /* Code flow error. Safety net. */
1224
1225         memset(&msg, 0, sizeof(msg));
1226         msg.cmd_type = cmd;
1227
1228         pthread_mutex_lock(consumer_sock->lock);
1229         DBG3("Sending %s command to consumer", command_name);
1230         ret = consumer_send_msg(consumer_sock, &msg);
1231         if (ret < 0) {
1232                 goto error;
1233         }
1234
1235         DBG3("Sending %s pipe %d to consumer on socket %d",
1236                         pipe_name,
1237                         pipe, *consumer_sock->fd_ptr);
1238         ret = consumer_send_fds(consumer_sock, &pipe, 1);
1239         if (ret < 0) {
1240                 goto error;
1241         }
1242
1243         DBG2("%s pipe successfully sent", pipe_name);
1244 error:
1245         pthread_mutex_unlock(consumer_sock->lock);
1246         return ret;
1247 }
1248
1249 int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock,
1250                 int pipe)
1251 {
1252         return consumer_send_pipe(consumer_sock,
1253                         LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe);
1254 }
1255
1256 /*
1257  * Ask the consumer if the data is pending for the specific session id.
1258  * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1259  */
1260 int consumer_is_data_pending(uint64_t session_id,
1261                 struct consumer_output *consumer)
1262 {
1263         int ret;
1264         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
1265         struct consumer_socket *socket;
1266         struct lttng_ht_iter iter;
1267         struct lttcomm_consumer_msg msg;
1268
1269         LTTNG_ASSERT(consumer);
1270
1271         DBG3("Consumer data pending for id %" PRIu64, session_id);
1272
1273         memset(&msg, 0, sizeof(msg));
1274         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1275         msg.u.data_pending.session_id = session_id;
1276
1277         /* Send command for each consumer */
1278         rcu_read_lock();
1279         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1280                         node.node) {
1281                 pthread_mutex_lock(socket->lock);
1282                 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1283                 if (ret < 0) {
1284                         pthread_mutex_unlock(socket->lock);
1285                         goto error_unlock;
1286                 }
1287
1288                 /*
1289                  * No need for a recv reply status because the answer to the command is
1290                  * the reply status message.
1291                  */
1292
1293                 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1294                 if (ret < 0) {
1295                         pthread_mutex_unlock(socket->lock);
1296                         goto error_unlock;
1297                 }
1298                 pthread_mutex_unlock(socket->lock);
1299
1300                 if (ret_code == 1) {
1301                         break;
1302                 }
1303         }
1304         rcu_read_unlock();
1305
1306         DBG("Consumer data is %s pending for session id %" PRIu64,
1307                         ret_code == 1 ? "" : "NOT", session_id);
1308         return ret_code;
1309
1310 error_unlock:
1311         rcu_read_unlock();
1312         return -1;
1313 }
1314
1315 /*
1316  * Send a flush command to consumer using the given channel key.
1317  *
1318  * Return 0 on success else a negative value.
1319  */
1320 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1321 {
1322         int ret;
1323         struct lttcomm_consumer_msg msg;
1324
1325         LTTNG_ASSERT(socket);
1326
1327         DBG2("Consumer flush channel key %" PRIu64, key);
1328
1329         memset(&msg, 0, sizeof(msg));
1330         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1331         msg.u.flush_channel.key = key;
1332
1333         pthread_mutex_lock(socket->lock);
1334         health_code_update();
1335
1336         ret = consumer_send_msg(socket, &msg);
1337         if (ret < 0) {
1338                 goto end;
1339         }
1340
1341 end:
1342         health_code_update();
1343         pthread_mutex_unlock(socket->lock);
1344         return ret;
1345 }
1346
1347 /*
1348  * Send a clear quiescent command to consumer using the given channel key.
1349  *
1350  * Return 0 on success else a negative value.
1351  */
1352 int consumer_clear_quiescent_channel(struct consumer_socket *socket, uint64_t key)
1353 {
1354         int ret;
1355         struct lttcomm_consumer_msg msg;
1356
1357         LTTNG_ASSERT(socket);
1358
1359         DBG2("Consumer clear quiescent channel key %" PRIu64, key);
1360
1361         memset(&msg, 0, sizeof(msg));
1362         msg.cmd_type = LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL;
1363         msg.u.clear_quiescent_channel.key = key;
1364
1365         pthread_mutex_lock(socket->lock);
1366         health_code_update();
1367
1368         ret = consumer_send_msg(socket, &msg);
1369         if (ret < 0) {
1370                 goto end;
1371         }
1372
1373 end:
1374         health_code_update();
1375         pthread_mutex_unlock(socket->lock);
1376         return ret;
1377 }
1378
1379 /*
1380  * Send a close metadata command to consumer using the given channel key.
1381  * Called with registry lock held.
1382  *
1383  * Return 0 on success else a negative value.
1384  */
1385 int consumer_close_metadata(struct consumer_socket *socket,
1386                 uint64_t metadata_key)
1387 {
1388         int ret;
1389         struct lttcomm_consumer_msg msg;
1390
1391         LTTNG_ASSERT(socket);
1392
1393         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1394
1395         memset(&msg, 0, sizeof(msg));
1396         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1397         msg.u.close_metadata.key = metadata_key;
1398
1399         pthread_mutex_lock(socket->lock);
1400         health_code_update();
1401
1402         ret = consumer_send_msg(socket, &msg);
1403         if (ret < 0) {
1404                 goto end;
1405         }
1406
1407 end:
1408         health_code_update();
1409         pthread_mutex_unlock(socket->lock);
1410         return ret;
1411 }
1412
1413 /*
1414  * Send a setup metdata command to consumer using the given channel key.
1415  *
1416  * Return 0 on success else a negative value.
1417  */
1418 int consumer_setup_metadata(struct consumer_socket *socket,
1419                 uint64_t metadata_key)
1420 {
1421         int ret;
1422         struct lttcomm_consumer_msg msg;
1423
1424         LTTNG_ASSERT(socket);
1425
1426         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1427
1428         memset(&msg, 0, sizeof(msg));
1429         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1430         msg.u.setup_metadata.key = metadata_key;
1431
1432         pthread_mutex_lock(socket->lock);
1433         health_code_update();
1434
1435         ret = consumer_send_msg(socket, &msg);
1436         if (ret < 0) {
1437                 goto end;
1438         }
1439
1440 end:
1441         health_code_update();
1442         pthread_mutex_unlock(socket->lock);
1443         return ret;
1444 }
1445
1446 /*
1447  * Send metadata string to consumer.
1448  * RCU read-side lock must be held to guarantee existence of socket.
1449  *
1450  * Return 0 on success else a negative value.
1451  */
1452 int consumer_push_metadata(struct consumer_socket *socket,
1453                 uint64_t metadata_key, char *metadata_str, size_t len,
1454                 size_t target_offset, uint64_t version)
1455 {
1456         int ret;
1457         struct lttcomm_consumer_msg msg;
1458
1459         LTTNG_ASSERT(socket);
1460
1461         DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1462
1463         pthread_mutex_lock(socket->lock);
1464
1465         memset(&msg, 0, sizeof(msg));
1466         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1467         msg.u.push_metadata.key = metadata_key;
1468         msg.u.push_metadata.target_offset = target_offset;
1469         msg.u.push_metadata.len = len;
1470         msg.u.push_metadata.version = version;
1471
1472         health_code_update();
1473         ret = consumer_send_msg(socket, &msg);
1474         if (ret < 0 || len == 0) {
1475                 goto end;
1476         }
1477
1478         DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1479                         len);
1480
1481         ret = consumer_socket_send(socket, metadata_str, len);
1482         if (ret < 0) {
1483                 goto end;
1484         }
1485
1486         health_code_update();
1487         ret = consumer_recv_status_reply(socket);
1488         if (ret < 0) {
1489                 goto end;
1490         }
1491
1492 end:
1493         pthread_mutex_unlock(socket->lock);
1494         health_code_update();
1495         return ret;
1496 }
1497
1498 /*
1499  * Ask the consumer to snapshot a specific channel using the key.
1500  *
1501  * Returns LTTNG_OK on success or else an LTTng error code.
1502  */
1503 enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket,
1504                 uint64_t key, const struct consumer_output *output, int metadata,
1505                 uid_t uid, gid_t gid, const char *channel_path, int wait,
1506                 uint64_t nb_packets_per_stream)
1507 {
1508         int ret;
1509         enum lttng_error_code status = LTTNG_OK;
1510         struct lttcomm_consumer_msg msg;
1511
1512         LTTNG_ASSERT(socket);
1513         LTTNG_ASSERT(output);
1514
1515         DBG("Consumer snapshot channel key %" PRIu64, key);
1516
1517         memset(&msg, 0, sizeof(msg));
1518         msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1519         msg.u.snapshot_channel.key = key;
1520         msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
1521         msg.u.snapshot_channel.metadata = metadata;
1522
1523         if (output->type == CONSUMER_DST_NET) {
1524                 msg.u.snapshot_channel.relayd_id =
1525                                 output->net_seq_index;
1526                 msg.u.snapshot_channel.use_relayd = 1;
1527         } else {
1528                 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1529         }
1530         ret = lttng_strncpy(msg.u.snapshot_channel.pathname,
1531                         channel_path,
1532                         sizeof(msg.u.snapshot_channel.pathname));
1533         if (ret < 0) {
1534                 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1535                                 sizeof(msg.u.snapshot_channel.pathname),
1536                                 strlen(channel_path),
1537                                 channel_path);
1538                 status = LTTNG_ERR_SNAPSHOT_FAIL;
1539                 goto error;
1540         }
1541
1542         health_code_update();
1543         pthread_mutex_lock(socket->lock);
1544         ret = consumer_send_msg(socket, &msg);
1545         pthread_mutex_unlock(socket->lock);
1546         if (ret < 0) {
1547                 switch (-ret) {
1548                 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1549                         status = LTTNG_ERR_CHAN_NOT_FOUND;
1550                         break;
1551                 default:
1552                         status = LTTNG_ERR_SNAPSHOT_FAIL;
1553                         break;
1554                 }
1555                 goto error;
1556         }
1557
1558 error:
1559         health_code_update();
1560         return status;
1561 }
1562
1563 /*
1564  * Ask the consumer the number of discarded events for a channel.
1565  */
1566 int consumer_get_discarded_events(uint64_t session_id, uint64_t channel_key,
1567                 struct consumer_output *consumer, uint64_t *discarded)
1568 {
1569         int ret;
1570         struct consumer_socket *socket;
1571         struct lttng_ht_iter iter;
1572         struct lttcomm_consumer_msg msg;
1573
1574         LTTNG_ASSERT(consumer);
1575
1576         DBG3("Consumer discarded events id %" PRIu64, session_id);
1577
1578         memset(&msg, 0, sizeof(msg));
1579         msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1580         msg.u.discarded_events.session_id = session_id;
1581         msg.u.discarded_events.channel_key = channel_key;
1582
1583         *discarded = 0;
1584
1585         /* Send command for each consumer */
1586         rcu_read_lock();
1587         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1588                         node.node) {
1589                 uint64_t consumer_discarded = 0;
1590                 pthread_mutex_lock(socket->lock);
1591                 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1592                 if (ret < 0) {
1593                         pthread_mutex_unlock(socket->lock);
1594                         goto end;
1595                 }
1596
1597                 /*
1598                  * No need for a recv reply status because the answer to the
1599                  * command is the reply status message.
1600                  */
1601                 ret = consumer_socket_recv(socket, &consumer_discarded,
1602                                 sizeof(consumer_discarded));
1603                 if (ret < 0) {
1604                         ERR("get discarded events");
1605                         pthread_mutex_unlock(socket->lock);
1606                         goto end;
1607                 }
1608                 pthread_mutex_unlock(socket->lock);
1609                 *discarded += consumer_discarded;
1610         }
1611         ret = 0;
1612         DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64,
1613                         *discarded, session_id);
1614
1615 end:
1616         rcu_read_unlock();
1617         return ret;
1618 }
1619
1620 /*
1621  * Ask the consumer the number of lost packets for a channel.
1622  */
1623 int consumer_get_lost_packets(uint64_t session_id, uint64_t channel_key,
1624                 struct consumer_output *consumer, uint64_t *lost)
1625 {
1626         int ret;
1627         struct consumer_socket *socket;
1628         struct lttng_ht_iter iter;
1629         struct lttcomm_consumer_msg msg;
1630
1631         LTTNG_ASSERT(consumer);
1632
1633         DBG3("Consumer lost packets id %" PRIu64, session_id);
1634
1635         memset(&msg, 0, sizeof(msg));
1636         msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1637         msg.u.lost_packets.session_id = session_id;
1638         msg.u.lost_packets.channel_key = channel_key;
1639
1640         *lost = 0;
1641
1642         /* Send command for each consumer */
1643         rcu_read_lock();
1644         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1645                         node.node) {
1646                 uint64_t consumer_lost = 0;
1647                 pthread_mutex_lock(socket->lock);
1648                 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1649                 if (ret < 0) {
1650                         pthread_mutex_unlock(socket->lock);
1651                         goto end;
1652                 }
1653
1654                 /*
1655                  * No need for a recv reply status because the answer to the
1656                  * command is the reply status message.
1657                  */
1658                 ret = consumer_socket_recv(socket, &consumer_lost,
1659                                 sizeof(consumer_lost));
1660                 if (ret < 0) {
1661                         ERR("get lost packets");
1662                         pthread_mutex_unlock(socket->lock);
1663                         goto end;
1664                 }
1665                 pthread_mutex_unlock(socket->lock);
1666                 *lost += consumer_lost;
1667         }
1668         ret = 0;
1669         DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64,
1670                         *lost, session_id);
1671
1672 end:
1673         rcu_read_unlock();
1674         return ret;
1675 }
1676
1677 /*
1678  * Ask the consumer to rotate a channel.
1679  *
1680  * The new_chunk_id is the session->rotate_count that has been incremented
1681  * when the rotation started. On the relay, this allows to keep track in which
1682  * chunk each stream is currently writing to (for the rotate_pending operation).
1683  */
1684 int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key,
1685                 uid_t uid, gid_t gid, struct consumer_output *output,
1686                 bool is_metadata_channel)
1687 {
1688         int ret;
1689         struct lttcomm_consumer_msg msg;
1690
1691         LTTNG_ASSERT(socket);
1692
1693         DBG("Consumer rotate channel key %" PRIu64, key);
1694
1695         pthread_mutex_lock(socket->lock);
1696         memset(&msg, 0, sizeof(msg));
1697         msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL;
1698         msg.u.rotate_channel.key = key;
1699         msg.u.rotate_channel.metadata = !!is_metadata_channel;
1700
1701         if (output->type == CONSUMER_DST_NET) {
1702                 msg.u.rotate_channel.relayd_id = output->net_seq_index;
1703         } else {
1704                 msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL;
1705         }
1706
1707         health_code_update();
1708         ret = consumer_send_msg(socket, &msg);
1709         if (ret < 0) {
1710                 switch (-ret) {
1711                 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1712                         ret = -LTTNG_ERR_CHAN_NOT_FOUND;
1713                         break;
1714                 default:
1715                         ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER;
1716                         break;
1717                 }
1718                 goto error;
1719         }
1720 error:
1721         pthread_mutex_unlock(socket->lock);
1722         health_code_update();
1723         return ret;
1724 }
1725
1726 int consumer_open_channel_packets(struct consumer_socket *socket, uint64_t key)
1727 {
1728         int ret;
1729         lttcomm_consumer_msg msg = {
1730                 .cmd_type = LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS,
1731         };
1732         msg.u.open_channel_packets.key = key;
1733
1734         LTTNG_ASSERT(socket);
1735
1736         DBG("Consumer open channel packets: channel key = %" PRIu64, key);
1737
1738         health_code_update();
1739
1740         pthread_mutex_lock(socket->lock);
1741         ret = consumer_send_msg(socket, &msg);
1742         pthread_mutex_unlock(socket->lock);
1743         if (ret < 0) {
1744                 goto error_socket;
1745         }
1746
1747 error_socket:
1748         health_code_update();
1749         return ret;
1750 }
1751
1752 int consumer_clear_channel(struct consumer_socket *socket, uint64_t key)
1753 {
1754         int ret;
1755         struct lttcomm_consumer_msg msg;
1756
1757         LTTNG_ASSERT(socket);
1758
1759         DBG("Consumer clear channel %" PRIu64, key);
1760
1761         memset(&msg, 0, sizeof(msg));
1762         msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL;
1763         msg.u.clear_channel.key = key;
1764
1765         health_code_update();
1766
1767         pthread_mutex_lock(socket->lock);
1768         ret = consumer_send_msg(socket, &msg);
1769         if (ret < 0) {
1770                 goto error_socket;
1771         }
1772
1773 error_socket:
1774         pthread_mutex_unlock(socket->lock);
1775
1776         health_code_update();
1777         return ret;
1778 }
1779
1780 int consumer_init(struct consumer_socket *socket,
1781                 const lttng_uuid sessiond_uuid)
1782 {
1783         int ret;
1784         struct lttcomm_consumer_msg msg = {
1785                 .cmd_type = LTTNG_CONSUMER_INIT,
1786         };
1787
1788         LTTNG_ASSERT(socket);
1789
1790         DBG("Sending consumer initialization command");
1791         lttng_uuid_copy(msg.u.init.sessiond_uuid, sessiond_uuid);
1792
1793         health_code_update();
1794         ret = consumer_send_msg(socket, &msg);
1795         if (ret < 0) {
1796                 goto error;
1797         }
1798
1799 error:
1800         health_code_update();
1801         return ret;
1802 }
1803
1804 /*
1805  * Ask the consumer to create a new chunk for a given session.
1806  *
1807  * Called with the consumer socket lock held.
1808  */
1809 int consumer_create_trace_chunk(struct consumer_socket *socket,
1810                 uint64_t relayd_id, uint64_t session_id,
1811                 struct lttng_trace_chunk *chunk,
1812                 const char *domain_subdir)
1813 {
1814         int ret;
1815         enum lttng_trace_chunk_status chunk_status;
1816         struct lttng_credentials chunk_credentials;
1817         const struct lttng_directory_handle *chunk_directory_handle = NULL;
1818         struct lttng_directory_handle *domain_handle = NULL;
1819         int domain_dirfd;
1820         const char *chunk_name;
1821         bool chunk_name_overridden;
1822         uint64_t chunk_id;
1823         time_t creation_timestamp;
1824         char creation_timestamp_buffer[ISO8601_STR_LEN];
1825         const char *creation_timestamp_str = "(none)";
1826         const bool chunk_has_local_output = relayd_id == -1ULL;
1827         enum lttng_trace_chunk_status tc_status;
1828         struct lttcomm_consumer_msg msg = {
1829                 .cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
1830         };
1831         msg.u.create_trace_chunk.session_id = session_id;
1832
1833         LTTNG_ASSERT(socket);
1834         LTTNG_ASSERT(chunk);
1835
1836         if (relayd_id != -1ULL) {
1837                 LTTNG_OPTIONAL_SET(&msg.u.create_trace_chunk.relayd_id,
1838                                 relayd_id);
1839         }
1840
1841         chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name,
1842                         &chunk_name_overridden);
1843         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1844                         chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1845                 ERR("Failed to get name of trace chunk");
1846                 ret = -LTTNG_ERR_FATAL;
1847                 goto error;
1848         }
1849         if (chunk_name_overridden) {
1850                 ret = lttng_strncpy(msg.u.create_trace_chunk.override_name,
1851                                 chunk_name,
1852                                 sizeof(msg.u.create_trace_chunk.override_name));
1853                 if (ret) {
1854                         ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1855                                         chunk_name);
1856                         ret = -LTTNG_ERR_FATAL;
1857                         goto error;
1858                 }
1859         }
1860
1861         chunk_status = lttng_trace_chunk_get_creation_timestamp(chunk,
1862                         &creation_timestamp);
1863         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1864                 ret = -LTTNG_ERR_FATAL;
1865                 goto error;
1866         }
1867         msg.u.create_trace_chunk.creation_timestamp =
1868                         (uint64_t) creation_timestamp;
1869         /* Only used for logging purposes. */
1870         ret = time_to_iso8601_str(creation_timestamp,
1871                         creation_timestamp_buffer,
1872                         sizeof(creation_timestamp_buffer));
1873         creation_timestamp_str = !ret ? creation_timestamp_buffer :
1874                         "(formatting error)";
1875
1876         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1877         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1878                 /*
1879                  * Anonymous trace chunks should never be transmitted
1880                  * to remote peers (consumerd and relayd). They are used
1881                  * internally for backward-compatibility purposes.
1882                  */
1883                 ret = -LTTNG_ERR_FATAL;
1884                 goto error;
1885         }
1886         msg.u.create_trace_chunk.chunk_id = chunk_id;
1887
1888         if (chunk_has_local_output) {
1889                 chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle(
1890                                 chunk, &chunk_directory_handle);
1891                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1892                         ret = -LTTNG_ERR_FATAL;
1893                         goto error;
1894                 }
1895                 chunk_status = lttng_trace_chunk_get_credentials(
1896                                 chunk, &chunk_credentials);
1897                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1898                         /*
1899                          * Not associating credentials to a sessiond chunk is a
1900                          * fatal internal error.
1901                          */
1902                         ret = -LTTNG_ERR_FATAL;
1903                         goto error;
1904                 }
1905                 tc_status = lttng_trace_chunk_create_subdirectory(
1906                                 chunk, domain_subdir);
1907                 if (tc_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1908                         PERROR("Failed to create chunk domain output directory \"%s\"",
1909                                 domain_subdir);
1910                         ret = -LTTNG_ERR_FATAL;
1911                         goto error;
1912                 }
1913                 domain_handle = lttng_directory_handle_create_from_handle(
1914                                 domain_subdir,
1915                                 chunk_directory_handle);
1916                 if (!domain_handle) {
1917                         ret = -LTTNG_ERR_FATAL;
1918                         goto error;
1919                 }
1920
1921                 /*
1922                  * This will only compile on platforms that support
1923                  * dirfd (POSIX.2008). This is fine as the session daemon
1924                  * is only built for such platforms.
1925                  *
1926                  * The ownership of the chunk directory handle's is maintained
1927                  * by the trace chunk.
1928                  */
1929                 domain_dirfd = lttng_directory_handle_get_dirfd(
1930                                 domain_handle);
1931                 LTTNG_ASSERT(domain_dirfd >= 0);
1932
1933                 msg.u.create_trace_chunk.credentials.value.uid =
1934                                 lttng_credentials_get_uid(&chunk_credentials);
1935                 msg.u.create_trace_chunk.credentials.value.gid =
1936                                 lttng_credentials_get_gid(&chunk_credentials);
1937                 msg.u.create_trace_chunk.credentials.is_set = 1;
1938         }
1939
1940         DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1941                         ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
1942                         ", creation_timestamp = %s",
1943                         relayd_id, session_id, chunk_id,
1944                         creation_timestamp_str);
1945         health_code_update();
1946         ret = consumer_send_msg(socket, &msg);
1947         health_code_update();
1948         if (ret < 0) {
1949                 ERR("Trace chunk creation error on consumer");
1950                 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1951                 goto error;
1952         }
1953
1954         if (chunk_has_local_output) {
1955                 DBG("Sending trace chunk domain directory fd to consumer");
1956                 health_code_update();
1957                 ret = consumer_send_fds(socket, &domain_dirfd, 1);
1958                 health_code_update();
1959                 if (ret < 0) {
1960                         ERR("Trace chunk creation error on consumer");
1961                         ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1962                         goto error;
1963                 }
1964         }
1965 error:
1966         lttng_directory_handle_put(domain_handle);
1967         return ret;
1968 }
1969
1970 /*
1971  * Ask the consumer to close a trace chunk for a given session.
1972  *
1973  * Called with the consumer socket lock held.
1974  */
1975 int consumer_close_trace_chunk(struct consumer_socket *socket,
1976                 uint64_t relayd_id, uint64_t session_id,
1977                 struct lttng_trace_chunk *chunk,
1978                 char *closed_trace_chunk_path)
1979 {
1980         int ret;
1981         enum lttng_trace_chunk_status chunk_status;
1982         lttcomm_consumer_msg msg = {
1983                 .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
1984         };
1985         msg.u.close_trace_chunk.session_id = session_id;
1986
1987         struct lttcomm_consumer_close_trace_chunk_reply reply;
1988         uint64_t chunk_id;
1989         time_t close_timestamp;
1990         enum lttng_trace_chunk_command_type close_command;
1991         const char *close_command_name = "none";
1992         struct lttng_dynamic_buffer path_reception_buffer;
1993
1994         LTTNG_ASSERT(socket);
1995         lttng_dynamic_buffer_init(&path_reception_buffer);
1996
1997         if (relayd_id != -1ULL) {
1998                 LTTNG_OPTIONAL_SET(
1999                                 &msg.u.close_trace_chunk.relayd_id, relayd_id);
2000         }
2001
2002         chunk_status = lttng_trace_chunk_get_close_command(
2003                         chunk, &close_command);
2004         switch (chunk_status) {
2005         case LTTNG_TRACE_CHUNK_STATUS_OK:
2006                 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command,
2007                                 (uint32_t) close_command);
2008                 break;
2009         case LTTNG_TRACE_CHUNK_STATUS_NONE:
2010                 break;
2011         default:
2012                 ERR("Failed to get trace chunk close command");
2013                 ret = -1;
2014                 goto error;
2015         }
2016
2017         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2018         /*
2019          * Anonymous trace chunks should never be transmitted to remote peers
2020          * (consumerd and relayd). They are used internally for
2021          * backward-compatibility purposes.
2022          */
2023         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2024         msg.u.close_trace_chunk.chunk_id = chunk_id;
2025
2026         chunk_status = lttng_trace_chunk_get_close_timestamp(chunk,
2027                         &close_timestamp);
2028         /*
2029          * A trace chunk should be closed locally before being closed remotely.
2030          * Otherwise, the close timestamp would never be transmitted to the
2031          * peers.
2032          */
2033         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2034         msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp;
2035
2036         if (msg.u.close_trace_chunk.close_command.is_set) {
2037                 close_command_name = lttng_trace_chunk_command_type_get_name(
2038                                 close_command);
2039         }
2040         DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
2041                         ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
2042                         ", close command = \"%s\"",
2043                         relayd_id, session_id, chunk_id, close_command_name);
2044
2045         health_code_update();
2046         ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg));
2047         if (ret < 0) {
2048                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2049                 goto error;
2050         }
2051         ret = consumer_socket_recv(socket, &reply, sizeof(reply));
2052         if (ret < 0) {
2053                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2054                 goto error;
2055         }
2056         if (reply.path_length >= LTTNG_PATH_MAX) {
2057                 ERR("Invalid path returned by relay daemon: %" PRIu32 "bytes exceeds maximal allowed length of %d bytes",
2058                                 reply.path_length, LTTNG_PATH_MAX);
2059                 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2060                 goto error;
2061         }
2062         ret = lttng_dynamic_buffer_set_size(&path_reception_buffer,
2063                         reply.path_length);
2064         if (ret) {
2065                 ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command");
2066                 ret = -LTTNG_ERR_NOMEM;
2067                 goto error;
2068         }
2069         ret = consumer_socket_recv(socket, path_reception_buffer.data,
2070                         path_reception_buffer.size);
2071         if (ret < 0) {
2072                 ERR("Communication error while receiving path of closed trace chunk");
2073                 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2074                 goto error;
2075         }
2076         if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') {
2077                 ERR("Invalid path returned by relay daemon: not null-terminated");
2078                 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2079                 goto error;
2080         }
2081         if (closed_trace_chunk_path) {
2082                 /*
2083                  * closed_trace_chunk_path is assumed to have a length >=
2084                  * LTTNG_PATH_MAX
2085                  */
2086                 memcpy(closed_trace_chunk_path, path_reception_buffer.data,
2087                                 path_reception_buffer.size);
2088         }
2089 error:
2090         lttng_dynamic_buffer_reset(&path_reception_buffer);
2091         health_code_update();
2092         return ret;
2093 }
2094
2095 /*
2096  * Ask the consumer if a trace chunk exists.
2097  *
2098  * Called with the consumer socket lock held.
2099  * Returns 0 on success, or a negative value on error.
2100  */
2101 int consumer_trace_chunk_exists(struct consumer_socket *socket,
2102                 uint64_t relayd_id, uint64_t session_id,
2103                 struct lttng_trace_chunk *chunk,
2104                 enum consumer_trace_chunk_exists_status *result)
2105 {
2106         int ret;
2107         enum lttng_trace_chunk_status chunk_status;
2108         lttcomm_consumer_msg msg = {
2109                 .cmd_type = LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,
2110         };
2111         msg.u.trace_chunk_exists.session_id = session_id;
2112
2113         uint64_t chunk_id;
2114         const char *consumer_reply_str;
2115
2116         LTTNG_ASSERT(socket);
2117
2118         if (relayd_id != -1ULL) {
2119                 LTTNG_OPTIONAL_SET(&msg.u.trace_chunk_exists.relayd_id,
2120                                 relayd_id);
2121         }
2122
2123         chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2124         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2125                 /*
2126                  * Anonymous trace chunks should never be transmitted
2127                  * to remote peers (consumerd and relayd). They are used
2128                  * internally for backward-compatibility purposes.
2129                  */
2130                 ret = -LTTNG_ERR_FATAL;
2131                 goto error;
2132         }
2133         msg.u.trace_chunk_exists.chunk_id = chunk_id;
2134
2135         DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
2136                         ", session_id = %" PRIu64
2137                         ", chunk_id = %" PRIu64, relayd_id, session_id, chunk_id);
2138
2139         health_code_update();
2140         ret = consumer_send_msg(socket, &msg);
2141         switch (-ret) {
2142         case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK:
2143                 consumer_reply_str = "unknown trace chunk";
2144                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK;
2145                 break;
2146         case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL:
2147                 consumer_reply_str = "trace chunk exists locally";
2148                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL;
2149                 break;
2150         case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE:
2151                 consumer_reply_str = "trace chunk exists on remote peer";
2152                 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE;
2153                 break;
2154         default:
2155                 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
2156                 ret = -1;
2157                 goto error;
2158         }
2159         DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
2160                         consumer_reply_str);
2161         ret = 0;
2162 error:
2163         health_code_update();
2164         return ret;
2165 }