src/bin/lttng-sessiond/consumer.c

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