src/bin/lttng-sessiond/consumer.c

   1 /*
   2  * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License, version 2 only, as
   6  * published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  11  * more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with
  14  * this program; if not, write to the Free Software Foundation, Inc., 51
  15  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _GNU_SOURCE
  19 #include <assert.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <sys/stat.h>
  24 #include <sys/types.h>
  25 #include <unistd.h>
  26 #include <inttypes.h>
  27
  28 #include <common/common.h>
  29 #include <common/defaults.h>
  30 #include <common/uri.h>
  31
  32 #include "consumer.h"
  33 #include "health.h"
  34 #include "ust-app.h"
  35 #include "utils.h"
  36
  37 /*
  38  * Receive a reply command status message from the consumer. Consumer socket
  39  * lock MUST be acquired before calling this function.
  40  *
  41  * Return 0 on success, -1 on recv error or a negative lttng error code which
  42  * was possibly returned by the consumer.
  43  */
  44 int consumer_recv_status_reply(struct consumer_socket *sock)
  45 {
  46         int ret;
  47         struct lttcomm_consumer_status_msg reply;
  48
  49         assert(sock);
  50
  51         ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
  52         if (ret <= 0) {
  53                 if (ret == 0) {
  54                         /* Orderly shutdown. Don't return 0 which means success. */
  55                         ret = -1;
  56                 }
  57                 /* The above call will print a PERROR on error. */
  58                 DBG("Fail to receive status reply on sock %d", sock->fd);
  59                 goto end;
  60         }
  61
  62         if (reply.ret_code == LTTNG_OK) {
  63                 /* All good. */
  64                 ret = 0;
  65         } else {
  66                 ret = -reply.ret_code;
  67                 DBG("Consumer ret code %d", ret);
  68         }
  69
  70 end:
  71         return ret;
  72 }
  73
  74 /*
  75  * Once the ASK_CHANNEL command is sent to the consumer, the channel
  76  * information are sent back. This call receives that data and populates key
  77  * and stream_count.
  78  *
  79  * On success return 0 and both key and stream_count are set. On error, a
  80  * negative value is sent back and both parameters are untouched.
  81  */
  82 int consumer_recv_status_channel(struct consumer_socket *sock,
  83                 uint64_t *key, unsigned int *stream_count)
  84 {
  85         int ret;
  86         struct lttcomm_consumer_status_channel reply;
  87
  88         assert(sock);
  89         assert(stream_count);
  90         assert(key);
  91
  92         ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
  93         if (ret <= 0) {
  94                 if (ret == 0) {
  95                         /* Orderly shutdown. Don't return 0 which means success. */
  96                         ret = -1;
  97                 }
  98                 /* The above call will print a PERROR on error. */
  99                 DBG("Fail to receive status reply on sock %d", sock->fd);
 100                 goto end;
 101         }
 102
 103         /* An error is possible so don't touch the key and stream_count. */
 104         if (reply.ret_code != LTTNG_OK) {
 105                 ret = -1;
 106                 goto end;
 107         }
 108
 109         *key = reply.key;
 110         *stream_count = reply.stream_count;
 111
 112 end:
 113         return ret;
 114 }
 115
 116 /*
 117  * Send destroy relayd command to consumer.
 118  *
 119  * On success return positive value. On error, negative value.
 120  */
 121 int consumer_send_destroy_relayd(struct consumer_socket *sock,
 122                 struct consumer_output *consumer)
 123 {
 124         int ret;
 125         struct lttcomm_consumer_msg msg;
 126
 127         assert(consumer);
 128         assert(sock);
 129
 130         DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
 131
 132         /* Bail out if consumer is disabled */
 133         if (!consumer->enabled) {
 134                 ret = LTTNG_OK;
 135                 DBG3("Consumer is disabled");
 136                 goto error;
 137         }
 138
 139         msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
 140         msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
 141
 142         pthread_mutex_lock(sock->lock);
 143         ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
 144         if (ret < 0) {
 145                 /* Indicate that the consumer is probably closing at this point. */
 146                 DBG("send consumer destroy relayd command");
 147                 goto error_send;
 148         }
 149
 150         /* Don't check the return value. The caller will do it. */
 151         ret = consumer_recv_status_reply(sock);
 152
 153         DBG2("Consumer send destroy relayd command done");
 154
 155 error_send:
 156         pthread_mutex_unlock(sock->lock);
 157 error:
 158         return ret;
 159 }
 160
 161 /*
 162  * For each consumer socket in the consumer output object, send a destroy
 163  * relayd command.
 164  */
 165 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
 166 {
 167         struct lttng_ht_iter iter;
 168         struct consumer_socket *socket;
 169
 170         assert(consumer);
 171
 172         /* Destroy any relayd connection */
 173         if (consumer->type == CONSUMER_DST_NET) {
 174                 rcu_read_lock();
 175                 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 176                                 node.node) {
 177                         int ret;
 178
 179                         /* Send destroy relayd command */
 180                         ret = consumer_send_destroy_relayd(socket, consumer);
 181                         if (ret < 0) {
 182                                 DBG("Unable to send destroy relayd command to consumer");
 183                                 /* Continue since we MUST delete everything at this point. */
 184                         }
 185                 }
 186                 rcu_read_unlock();
 187         }
 188 }
 189
 190 /*
 191  * From a consumer_data structure, allocate and add a consumer socket to the
 192  * consumer output.
 193  *
 194  * Return 0 on success, else negative value on error
 195  */
 196 int consumer_create_socket(struct consumer_data *data,
 197                 struct consumer_output *output)
 198 {
 199         int ret = 0;
 200         struct consumer_socket *socket;
 201
 202         assert(data);
 203
 204         if (output == NULL || data->cmd_sock < 0) {
 205                 /*
 206                  * Not an error. Possible there is simply not spawned consumer or it's
 207                  * disabled for the tracing session asking the socket.
 208                  */
 209                 goto error;
 210         }
 211
 212         rcu_read_lock();
 213         socket = consumer_find_socket(data->cmd_sock, output);
 214         rcu_read_unlock();
 215         if (socket == NULL) {
 216                 socket = consumer_allocate_socket(data->cmd_sock);
 217                 if (socket == NULL) {
 218                         ret = -1;
 219                         goto error;
 220                 }
 221
 222                 socket->registered = 0;
 223                 socket->lock = &data->lock;
 224                 rcu_read_lock();
 225                 consumer_add_socket(socket, output);
 226                 rcu_read_unlock();
 227         }
 228
 229         socket->type = data->type;
 230
 231         DBG3("Consumer socket created (fd: %d) and added to output",
 232                         data->cmd_sock);
 233
 234 error:
 235         return ret;
 236 }
 237
 238 /*
 239  * Return the consumer socket from the given consumer output with the right
 240  * bitness. On error, returns NULL.
 241  *
 242  * The caller MUST acquire a rcu read side lock and keep it until the socket
 243  * object reference is not needed anymore.
 244  */
 245 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
 246                 struct consumer_output *consumer)
 247 {
 248         int consumer_fd;
 249         struct consumer_socket *socket = NULL;
 250
 251         switch (bits) {
 252         case 64:
 253                 consumer_fd = uatomic_read(&ust_consumerd64_fd);
 254                 break;
 255         case 32:
 256                 consumer_fd = uatomic_read(&ust_consumerd32_fd);
 257                 break;
 258         default:
 259                 assert(0);
 260                 goto end;
 261         }
 262
 263         socket = consumer_find_socket(consumer_fd, consumer);
 264         if (!socket) {
 265                 ERR("Consumer socket fd %d not found in consumer obj %p",
 266                                 consumer_fd, consumer);
 267         }
 268
 269 end:
 270         return socket;
 271 }
 272
 273 /*
 274  * Find a consumer_socket in a consumer_output hashtable. Read side lock must
 275  * be acquired before calling this function and across use of the
 276  * returned consumer_socket.
 277  */
 278 struct consumer_socket *consumer_find_socket(int key,
 279                 struct consumer_output *consumer)
 280 {
 281         struct lttng_ht_iter iter;
 282         struct lttng_ht_node_ulong *node;
 283         struct consumer_socket *socket = NULL;
 284
 285         /* Negative keys are lookup failures */
 286         if (key < 0 || consumer == NULL) {
 287                 return NULL;
 288         }
 289
 290         lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
 291                         &iter);
 292         node = lttng_ht_iter_get_node_ulong(&iter);
 293         if (node != NULL) {
 294                 socket = caa_container_of(node, struct consumer_socket, node);
 295         }
 296
 297         return socket;
 298 }
 299
 300 /*
 301  * Allocate a new consumer_socket and return the pointer.
 302  */
 303 struct consumer_socket *consumer_allocate_socket(int fd)
 304 {
 305         struct consumer_socket *socket = NULL;
 306
 307         socket = zmalloc(sizeof(struct consumer_socket));
 308         if (socket == NULL) {
 309                 PERROR("zmalloc consumer socket");
 310                 goto error;
 311         }
 312
 313         socket->fd = fd;
 314         lttng_ht_node_init_ulong(&socket->node, fd);
 315
 316 error:
 317         return socket;
 318 }
 319
 320 /*
 321  * Add consumer socket to consumer output object. Read side lock must be
 322  * acquired before calling this function.
 323  */
 324 void consumer_add_socket(struct consumer_socket *sock,
 325                 struct consumer_output *consumer)
 326 {
 327         assert(sock);
 328         assert(consumer);
 329
 330         lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
 331 }
 332
 333 /*
 334  * Delte consumer socket to consumer output object. Read side lock must be
 335  * acquired before calling this function.
 336  */
 337 void consumer_del_socket(struct consumer_socket *sock,
 338                 struct consumer_output *consumer)
 339 {
 340         int ret;
 341         struct lttng_ht_iter iter;
 342
 343         assert(sock);
 344         assert(consumer);
 345
 346         iter.iter.node = &sock->node.node;
 347         ret = lttng_ht_del(consumer->socks, &iter);
 348         assert(!ret);
 349 }
 350
 351 /*
 352  * RCU destroy call function.
 353  */
 354 static void destroy_socket_rcu(struct rcu_head *head)
 355 {
 356         struct lttng_ht_node_ulong *node =
 357                 caa_container_of(head, struct lttng_ht_node_ulong, head);
 358         struct consumer_socket *socket =
 359                 caa_container_of(node, struct consumer_socket, node);
 360
 361         free(socket);
 362 }
 363
 364 /*
 365  * Destroy and free socket pointer in a call RCU. Read side lock must be
 366  * acquired before calling this function.
 367  */
 368 void consumer_destroy_socket(struct consumer_socket *sock)
 369 {
 370         assert(sock);
 371
 372         /*
 373          * We DO NOT close the file descriptor here since it is global to the
 374          * session daemon and is closed only if the consumer dies or a custom
 375          * consumer was registered,
 376          */
 377         if (sock->registered) {
 378                 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
 379                 lttcomm_close_unix_sock(sock->fd);
 380         }
 381
 382         call_rcu(&sock->node.head, destroy_socket_rcu);
 383 }
 384
 385 /*
 386  * Allocate and assign data to a consumer_output object.
 387  *
 388  * Return pointer to structure.
 389  */
 390 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
 391 {
 392         struct consumer_output *output = NULL;
 393
 394         output = zmalloc(sizeof(struct consumer_output));
 395         if (output == NULL) {
 396                 PERROR("zmalloc consumer_output");
 397                 goto error;
 398         }
 399
 400         /* By default, consumer output is enabled */
 401         output->enabled = 1;
 402         output->type = type;
 403         output->net_seq_index = (uint64_t) -1ULL;
 404
 405         output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 406
 407 error:
 408         return output;
 409 }
 410
 411 /*
 412  * Delete the consumer_output object from the list and free the ptr.
 413  *
 414  * Should *NOT* be called with RCU read-side lock held.
 415  */
 416 void consumer_destroy_output(struct consumer_output *obj)
 417 {
 418         if (obj == NULL) {
 419                 return;
 420         }
 421
 422         if (obj->socks) {
 423                 struct lttng_ht_iter iter;
 424                 struct consumer_socket *socket;
 425
 426                 rcu_read_lock();
 427                 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 428                         consumer_del_socket(socket, obj);
 429                         consumer_destroy_socket(socket);
 430                 }
 431                 rcu_read_unlock();
 432
 433                 /* Finally destroy HT */
 434                 ht_cleanup_push(obj->socks);
 435         }
 436
 437         free(obj);
 438 }
 439
 440 /*
 441  * Copy consumer output and returned the newly allocated copy.
 442  *
 443  * Should *NOT* be called with RCU read-side lock held.
 444  */
 445 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
 446 {
 447         int ret;
 448         struct lttng_ht *tmp_ht_ptr;
 449         struct consumer_output *output;
 450
 451         assert(obj);
 452
 453         output = consumer_create_output(obj->type);
 454         if (output == NULL) {
 455                 goto error;
 456         }
 457         /* Avoid losing the HT reference after the memcpy() */
 458         tmp_ht_ptr = output->socks;
 459
 460         memcpy(output, obj, sizeof(struct consumer_output));
 461
 462         /* Putting back the HT pointer and start copying socket(s). */
 463         output->socks = tmp_ht_ptr;
 464
 465         ret = consumer_copy_sockets(output, obj);
 466         if (ret < 0) {
 467                 goto malloc_error;
 468         }
 469
 470 error:
 471         return output;
 472
 473 malloc_error:
 474         consumer_destroy_output(output);
 475         return NULL;
 476 }
 477
 478 /*
 479  * Copy consumer sockets from src to dst.
 480  *
 481  * Return 0 on success or else a negative value.
 482  */
 483 int consumer_copy_sockets(struct consumer_output *dst,
 484                 struct consumer_output *src)
 485 {
 486         int ret = 0;
 487         struct lttng_ht_iter iter;
 488         struct consumer_socket *socket, *copy_sock;
 489
 490         assert(dst);
 491         assert(src);
 492
 493         rcu_read_lock();
 494         cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
 495                 /* Ignore socket that are already there. */
 496                 copy_sock = consumer_find_socket(socket->fd, dst);
 497                 if (copy_sock) {
 498                         continue;
 499                 }
 500
 501                 /* Create new socket object. */
 502                 copy_sock = consumer_allocate_socket(socket->fd);
 503                 if (copy_sock == NULL) {
 504                         rcu_read_unlock();
 505                         ret = -ENOMEM;
 506                         goto error;
 507                 }
 508
 509                 copy_sock->registered = socket->registered;
 510                 /*
 511                  * This is valid because this lock is shared accross all consumer
 512                  * object being the global lock of the consumer data structure of the
 513                  * session daemon.
 514                  */
 515                 copy_sock->lock = socket->lock;
 516                 consumer_add_socket(copy_sock, dst);
 517         }
 518         rcu_read_unlock();
 519
 520 error:
 521         return ret;
 522 }
 523
 524 /*
 525  * Set network URI to the consumer output object.
 526  *
 527  * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
 528  * error.
 529  */
 530 int consumer_set_network_uri(struct consumer_output *obj,
 531                 struct lttng_uri *uri)
 532 {
 533         int ret;
 534         char tmp_path[PATH_MAX];
 535         char hostname[HOST_NAME_MAX];
 536         struct lttng_uri *dst_uri = NULL;
 537
 538         /* Code flow error safety net. */
 539         assert(obj);
 540         assert(uri);
 541
 542         switch (uri->stype) {
 543         case LTTNG_STREAM_CONTROL:
 544                 dst_uri = &obj->dst.net.control;
 545                 obj->dst.net.control_isset = 1;
 546                 if (uri->port == 0) {
 547                         /* Assign default port. */
 548                         uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 549                 } else {
 550                         if (obj->dst.net.data_isset && uri->port ==
 551                                         obj->dst.net.data.port) {
 552                                 ret = -LTTNG_ERR_INVALID;
 553                                 goto error;
 554                         }
 555                 }
 556                 DBG3("Consumer control URI set with port %d", uri->port);
 557                 break;
 558         case LTTNG_STREAM_DATA:
 559                 dst_uri = &obj->dst.net.data;
 560                 obj->dst.net.data_isset = 1;
 561                 if (uri->port == 0) {
 562                         /* Assign default port. */
 563                         uri->port = DEFAULT_NETWORK_DATA_PORT;
 564                 } else {
 565                         if (obj->dst.net.control_isset && uri->port ==
 566                                         obj->dst.net.control.port) {
 567                                 ret = -LTTNG_ERR_INVALID;
 568                                 goto error;
 569                         }
 570                 }
 571                 DBG3("Consumer data URI set with port %d", uri->port);
 572                 break;
 573         default:
 574                 ERR("Set network uri type unknown %d", uri->stype);
 575                 ret = -LTTNG_ERR_INVALID;
 576                 goto error;
 577         }
 578
 579         ret = uri_compare(dst_uri, uri);
 580         if (!ret) {
 581                 /* Same URI, don't touch it and return success. */
 582                 DBG3("URI network compare are the same");
 583                 goto equal;
 584         }
 585
 586         /* URIs were not equal, replacing it. */
 587         memset(dst_uri, 0, sizeof(struct lttng_uri));
 588         memcpy(dst_uri, uri, sizeof(struct lttng_uri));
 589         obj->type = CONSUMER_DST_NET;
 590
 591         /* Handle subdir and add hostname in front. */
 592         if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
 593                 /* Get hostname to append it in the pathname */
 594                 ret = gethostname(hostname, sizeof(hostname));
 595                 if (ret < 0) {
 596                         PERROR("gethostname. Fallback on default localhost");
 597                         strncpy(hostname, "localhost", sizeof(hostname));
 598                 }
 599                 hostname[sizeof(hostname) - 1] = '\0';
 600
 601                 /* Setup consumer subdir if none present in the control URI */
 602                 if (strlen(dst_uri->subdir) == 0) {
 603                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 604                                         hostname, obj->subdir);
 605                 } else {
 606                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 607                                         hostname, dst_uri->subdir);
 608                 }
 609                 if (ret < 0) {
 610                         PERROR("snprintf set consumer uri subdir");
 611                         ret = -LTTNG_ERR_NOMEM;
 612                         goto error;
 613                 }
 614
 615                 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
 616                 DBG3("Consumer set network uri subdir path %s", tmp_path);
 617         }
 618
 619         return 0;
 620 equal:
 621         return 1;
 622 error:
 623         return ret;
 624 }
 625
 626 /*
 627  * Send file descriptor to consumer via sock.
 628  */
 629 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
 630 {
 631         int ret;
 632
 633         assert(fds);
 634         assert(sock);
 635         assert(nb_fd > 0);
 636
 637         ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
 638         if (ret < 0) {
 639                 /* The above call will print a PERROR on error. */
 640                 DBG("Error when sending consumer fds on sock %d", sock->fd);
 641                 goto error;
 642         }
 643
 644         ret = consumer_recv_status_reply(sock);
 645
 646 error:
 647         return ret;
 648 }
 649
 650 /*
 651  * Consumer send communication message structure to consumer.
 652  */
 653 int consumer_send_msg(struct consumer_socket *sock,
 654                 struct lttcomm_consumer_msg *msg)
 655 {
 656         int ret;
 657
 658         assert(msg);
 659         assert(sock);
 660         assert(sock->fd >= 0);
 661
 662         ret = lttcomm_send_unix_sock(sock->fd, msg,
 663                         sizeof(struct lttcomm_consumer_msg));
 664         if (ret < 0) {
 665                 /* The above call will print a PERROR on error. */
 666                 DBG("Error when sending consumer channel on sock %d", sock->fd);
 667                 goto error;
 668         }
 669
 670         ret = consumer_recv_status_reply(sock);
 671
 672 error:
 673         return ret;
 674 }
 675
 676 /*
 677  * Consumer send channel communication message structure to consumer.
 678  */
 679 int consumer_send_channel(struct consumer_socket *sock,
 680                 struct lttcomm_consumer_msg *msg)
 681 {
 682         int ret;
 683
 684         assert(msg);
 685         assert(sock);
 686         assert(sock->fd >= 0);
 687
 688         ret = lttcomm_send_unix_sock(sock->fd, msg,
 689                         sizeof(struct lttcomm_consumer_msg));
 690         if (ret < 0) {
 691                 /* The above call will print a PERROR on error. */
 692                 DBG("Error when sending consumer channel on sock %d", sock->fd);
 693                 goto error;
 694         }
 695
 696         ret = consumer_recv_status_reply(sock);
 697
 698 error:
 699         return ret;
 700 }
 701
 702 /*
 703  * Populate the given consumer msg structure with the ask_channel command
 704  * information.
 705  */
 706 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 707                 uint64_t subbuf_size,
 708                 uint64_t num_subbuf,
 709                 int overwrite,
 710                 unsigned int switch_timer_interval,
 711                 unsigned int read_timer_interval,
 712                 int output,
 713                 int type,
 714                 uint64_t session_id,
 715                 const char *pathname,
 716                 const char *name,
 717                 uid_t uid,
 718                 gid_t gid,
 719                 uint64_t relayd_id,
 720                 uint64_t key,
 721                 unsigned char *uuid,
 722                 uint32_t chan_id,
 723                 uint64_t tracefile_size,
 724                 uint64_t tracefile_count)
 725 {
 726         assert(msg);
 727
 728         /* Zeroed structure */
 729         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 730
 731         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 732         msg->u.ask_channel.subbuf_size = subbuf_size;
 733         msg->u.ask_channel.num_subbuf = num_subbuf ;
 734         msg->u.ask_channel.overwrite = overwrite;
 735         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 736         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 737         msg->u.ask_channel.output = output;
 738         msg->u.ask_channel.type = type;
 739         msg->u.ask_channel.session_id = session_id;
 740         msg->u.ask_channel.uid = uid;
 741         msg->u.ask_channel.gid = gid;
 742         msg->u.ask_channel.relayd_id = relayd_id;
 743         msg->u.ask_channel.key = key;
 744         msg->u.ask_channel.chan_id = chan_id;
 745         msg->u.ask_channel.tracefile_size = tracefile_size;
 746         msg->u.ask_channel.tracefile_count = tracefile_count;
 747
 748         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 749
 750         strncpy(msg->u.ask_channel.pathname, pathname,
 751                         sizeof(msg->u.ask_channel.pathname));
 752         msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 753
 754         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 755         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 756 }
 757
 758 /*
 759  * Init channel communication message structure.
 760  */
 761 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 762                 enum lttng_consumer_command cmd,
 763                 uint64_t channel_key,
 764                 uint64_t session_id,
 765                 const char *pathname,
 766                 uid_t uid,
 767                 gid_t gid,
 768                 uint64_t relayd_id,
 769                 const char *name,
 770                 unsigned int nb_init_streams,
 771                 enum lttng_event_output output,
 772                 int type,
 773                 uint64_t tracefile_size,
 774                 uint64_t tracefile_count)
 775 {
 776         assert(msg);
 777
 778         /* Zeroed structure */
 779         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 780
 781         /* Send channel */
 782         msg->cmd_type = cmd;
 783         msg->u.channel.channel_key = channel_key;
 784         msg->u.channel.session_id = session_id;
 785         msg->u.channel.uid = uid;
 786         msg->u.channel.gid = gid;
 787         msg->u.channel.relayd_id = relayd_id;
 788         msg->u.channel.nb_init_streams = nb_init_streams;
 789         msg->u.channel.output = output;
 790         msg->u.channel.type = type;
 791         msg->u.channel.tracefile_size = tracefile_size;
 792         msg->u.channel.tracefile_count = tracefile_count;
 793
 794         strncpy(msg->u.channel.pathname, pathname,
 795                         sizeof(msg->u.channel.pathname));
 796         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
 797
 798         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
 799         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 800 }
 801
 802 /*
 803  * Init stream communication message structure.
 804  */
 805 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 806                 enum lttng_consumer_command cmd,
 807                 uint64_t channel_key,
 808                 uint64_t stream_key,
 809                 int cpu)
 810 {
 811         assert(msg);
 812
 813         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 814
 815         msg->cmd_type = cmd;
 816         msg->u.stream.channel_key = channel_key;
 817         msg->u.stream.stream_key = stream_key;
 818         msg->u.stream.cpu = cpu;
 819 }
 820
 821 /*
 822  * Send stream communication structure to the consumer.
 823  */
 824 int consumer_send_stream(struct consumer_socket *sock,
 825                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
 826                 int *fds, size_t nb_fd)
 827 {
 828         int ret;
 829
 830         assert(msg);
 831         assert(dst);
 832         assert(sock);
 833         assert(fds);
 834
 835         /* Send on socket */
 836         ret = lttcomm_send_unix_sock(sock->fd, msg,
 837                         sizeof(struct lttcomm_consumer_msg));
 838         if (ret < 0) {
 839                 /* The above call will print a PERROR on error. */
 840                 DBG("Error when sending consumer stream on sock %d", sock->fd);
 841                 goto error;
 842         }
 843
 844         ret = consumer_recv_status_reply(sock);
 845         if (ret < 0) {
 846                 goto error;
 847         }
 848
 849         ret = consumer_send_fds(sock, fds, nb_fd);
 850         if (ret < 0) {
 851                 goto error;
 852         }
 853
 854 error:
 855         return ret;
 856 }
 857
 858 /*
 859  * Send relayd socket to consumer associated with a session name.
 860  *
 861  * On success return positive value. On error, negative value.
 862  */
 863 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 864                 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
 865                 enum lttng_stream_type type, uint64_t session_id)
 866 {
 867         int ret;
 868         struct lttcomm_consumer_msg msg;
 869
 870         /* Code flow error. Safety net. */
 871         assert(rsock);
 872         assert(consumer);
 873         assert(consumer_sock);
 874
 875         /* Bail out if consumer is disabled */
 876         if (!consumer->enabled) {
 877                 ret = LTTNG_OK;
 878                 goto error;
 879         }
 880
 881         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 882         /*
 883          * Assign network consumer output index using the temporary consumer since
 884          * this call should only be made from within a set_consumer_uri() function
 885          * call in the session daemon.
 886          */
 887         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 888         msg.u.relayd_sock.type = type;
 889         msg.u.relayd_sock.session_id = session_id;
 890         memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
 891
 892         DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
 893         ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
 894         if (ret < 0) {
 895                 /* The above call will print a PERROR on error. */
 896                 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
 897                 goto error;
 898         }
 899
 900         ret = consumer_recv_status_reply(consumer_sock);
 901         if (ret < 0) {
 902                 goto error;
 903         }
 904
 905         DBG3("Sending relayd socket file descriptor to consumer");
 906         ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
 907         if (ret < 0) {
 908                 goto error;
 909         }
 910
 911         DBG2("Consumer relayd socket sent");
 912
 913 error:
 914         return ret;
 915 }
 916
 917 /*
 918  * Set consumer subdirectory using the session name and a generated datetime if
 919  * needed. This is appended to the current subdirectory.
 920  */
 921 int consumer_set_subdir(struct consumer_output *consumer,
 922                 const char *session_name)
 923 {
 924         int ret = 0;
 925         unsigned int have_default_name = 0;
 926         char datetime[16], tmp_path[PATH_MAX];
 927         time_t rawtime;
 928         struct tm *timeinfo;
 929
 930         assert(consumer);
 931         assert(session_name);
 932
 933         memset(tmp_path, 0, sizeof(tmp_path));
 934
 935         /* Flag if we have a default session. */
 936         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
 937                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
 938                 have_default_name = 1;
 939         } else {
 940                 /* Get date and time for session path */
 941                 time(&rawtime);
 942                 timeinfo = localtime(&rawtime);
 943                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
 944         }
 945
 946         if (have_default_name) {
 947                 ret = snprintf(tmp_path, sizeof(tmp_path),
 948                                 "%s/%s", consumer->subdir, session_name);
 949         } else {
 950                 ret = snprintf(tmp_path, sizeof(tmp_path),
 951                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
 952         }
 953         if (ret < 0) {
 954                 PERROR("snprintf session name date");
 955                 goto error;
 956         }
 957
 958         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
 959         DBG2("Consumer subdir set to %s", consumer->subdir);
 960
 961 error:
 962         return ret;
 963 }
 964
 965 /*
 966  * Ask the consumer if the data is ready to read (NOT pending) for the specific
 967  * session id.
 968  *
 969  * This function has a different behavior with the consumer i.e. that it waits
 970  * for a reply from the consumer if yes or no the data is pending.
 971  */
 972 int consumer_is_data_pending(uint64_t session_id,
 973                 struct consumer_output *consumer)
 974 {
 975         int ret;
 976         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
 977         struct consumer_socket *socket;
 978         struct lttng_ht_iter iter;
 979         struct lttcomm_consumer_msg msg;
 980
 981         assert(consumer);
 982
 983         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
 984
 985         msg.u.data_pending.session_id = session_id;
 986
 987         DBG3("Consumer data pending for id %" PRIu64, session_id);
 988
 989         /* Send command for each consumer */
 990         rcu_read_lock();
 991         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 992                         node.node) {
 993                 /* Code flow error */
 994                 assert(socket->fd >= 0);
 995
 996                 pthread_mutex_lock(socket->lock);
 997
 998                 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
 999                 if (ret < 0) {
1000                         /* The above call will print a PERROR on error. */
1001                         DBG("Error on consumer is data pending on sock %d", socket->fd);
1002                         pthread_mutex_unlock(socket->lock);
1003                         goto error_unlock;
1004                 }
1005
1006                 /*
1007                  * No need for a recv reply status because the answer to the command is
1008                  * the reply status message.
1009                  */
1010
1011                 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
1012                 if (ret <= 0) {
1013                         if (ret == 0) {
1014                                 /* Orderly shutdown. Don't return 0 which means success. */
1015                                 ret = -1;
1016                         }
1017                         /* The above call will print a PERROR on error. */
1018                         DBG("Error on recv consumer is data pending on sock %d", socket->fd);
1019                         pthread_mutex_unlock(socket->lock);
1020                         goto error_unlock;
1021                 }
1022
1023                 pthread_mutex_unlock(socket->lock);
1024
1025                 if (ret_code == 1) {
1026                         break;
1027                 }
1028         }
1029         rcu_read_unlock();
1030
1031         DBG("Consumer data is %s pending for session id %" PRIu64,
1032                         ret_code == 1 ? "" : "NOT", session_id);
1033         return ret_code;
1034
1035 error_unlock:
1036         rcu_read_unlock();
1037         return -1;
1038 }
1039
1040 /*
1041  * Send a flush command to consumer using the given channel key.
1042  *
1043  * Return 0 on success else a negative value.
1044  */
1045 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1046 {
1047         int ret;
1048         struct lttcomm_consumer_msg msg;
1049
1050         assert(socket);
1051         assert(socket->fd >= 0);
1052
1053         DBG2("Consumer flush channel key %" PRIu64, key);
1054
1055         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1056         msg.u.flush_channel.key = key;
1057
1058         pthread_mutex_lock(socket->lock);
1059         health_code_update();
1060
1061         ret = consumer_send_msg(socket, &msg);
1062         if (ret < 0) {
1063                 goto end;
1064         }
1065
1066 end:
1067         health_code_update();
1068         pthread_mutex_unlock(socket->lock);
1069         return ret;
1070 }
1071
1072 /*
1073  * Send a close metdata command to consumer using the given channel key.
1074  *
1075  * Return 0 on success else a negative value.
1076  */
1077 int consumer_close_metadata(struct consumer_socket *socket,
1078                 uint64_t metadata_key)
1079 {
1080         int ret;
1081         struct lttcomm_consumer_msg msg;
1082
1083         assert(socket);
1084         assert(socket->fd >= 0);
1085
1086         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1087
1088         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1089         msg.u.close_metadata.key = metadata_key;
1090
1091         pthread_mutex_lock(socket->lock);
1092         health_code_update();
1093
1094         ret = consumer_send_msg(socket, &msg);
1095         if (ret < 0) {
1096                 goto end;
1097         }
1098
1099 end:
1100         health_code_update();
1101         pthread_mutex_unlock(socket->lock);
1102         return ret;
1103 }
1104
1105 /*
1106  * Send a setup metdata command to consumer using the given channel key.
1107  *
1108  * Return 0 on success else a negative value.
1109  */
1110 int consumer_setup_metadata(struct consumer_socket *socket,
1111                 uint64_t metadata_key)
1112 {
1113         int ret;
1114         struct lttcomm_consumer_msg msg;
1115
1116         assert(socket);
1117         assert(socket->fd >= 0);
1118
1119         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1120
1121         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1122         msg.u.setup_metadata.key = metadata_key;
1123
1124         pthread_mutex_lock(socket->lock);
1125         health_code_update();
1126
1127         ret = consumer_send_msg(socket, &msg);
1128         if (ret < 0) {
1129                 goto end;
1130         }
1131
1132 end:
1133         health_code_update();
1134         pthread_mutex_unlock(socket->lock);
1135         return ret;
1136 }
1137
1138 /*
1139  * Send metadata string to consumer. Socket lock MUST be acquired.
1140  *
1141  * Return 0 on success else a negative value.
1142  */
1143 int consumer_push_metadata(struct consumer_socket *socket,
1144                 uint64_t metadata_key, char *metadata_str, size_t len,
1145                 size_t target_offset)
1146 {
1147         int ret;
1148         struct lttcomm_consumer_msg msg;
1149
1150         assert(socket);
1151         assert(socket->fd >= 0);
1152
1153         DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1154
1155         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1156         msg.u.push_metadata.key = metadata_key;
1157         msg.u.push_metadata.target_offset = target_offset;
1158         msg.u.push_metadata.len = len;
1159
1160         health_code_update();
1161         ret = consumer_send_msg(socket, &msg);
1162         if (ret < 0 || len == 0) {
1163                 goto end;
1164         }
1165
1166         DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
1167
1168         ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1169         if (ret < 0) {
1170                 goto end;
1171         }
1172
1173         health_code_update();
1174         ret = consumer_recv_status_reply(socket);
1175         if (ret < 0) {
1176                 goto end;
1177         }
1178
1179 end:
1180         health_code_update();
1181         return ret;
1182 }
1183
1184 /*
1185  * Ask the consumer to snapshot a specific channel using the key.
1186  *
1187  * Return 0 on success or else a negative error.
1188  */
1189 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1190                 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1191                 int wait)
1192 {
1193         int ret;
1194         struct lttcomm_consumer_msg msg;
1195
1196         assert(socket);
1197         assert(socket->fd >= 0);
1198         assert(output);
1199         assert(output->consumer);
1200
1201         DBG("Consumer snapshot channel key %" PRIu64, key);
1202
1203         memset(&msg, 0, sizeof(msg));
1204
1205         msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1206         msg.u.snapshot_channel.key = key;
1207         msg.u.snapshot_channel.max_size = output->max_size;
1208         msg.u.snapshot_channel.metadata = metadata;
1209
1210         if (output->consumer->type == CONSUMER_DST_NET) {
1211                 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1212                 msg.u.snapshot_channel.use_relayd = 1;
1213                 ret = snprintf(msg.u.snapshot_channel.pathname,
1214                                 sizeof(msg.u.snapshot_channel.pathname), "%s/%s",
1215                                 output->consumer->subdir, DEFAULT_SNAPSHOT_NAME);
1216                 if (ret < 0) {
1217                         ret = -LTTNG_ERR_NOMEM;
1218                         goto error;
1219                 }
1220         } else {
1221                 ret = snprintf(msg.u.snapshot_channel.pathname,
1222                                 sizeof(msg.u.snapshot_channel.pathname), "%s/%s",
1223                                 output->consumer->dst.trace_path, DEFAULT_SNAPSHOT_NAME);
1224                 if (ret < 0) {
1225                         ret = -LTTNG_ERR_NOMEM;
1226                         goto error;
1227                 }
1228
1229                 /* Create directory. Ignore if exist. */
1230                 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1231                                 S_IRWXU | S_IRWXG, uid, gid);
1232                 if (ret < 0) {
1233                         if (ret != -EEXIST) {
1234                                 ERR("Trace directory creation error");
1235                                 goto error;
1236                         }
1237                 }
1238         }
1239
1240         health_code_update();
1241         ret = consumer_send_msg(socket, &msg);
1242         if (ret < 0) {
1243                 goto error;
1244         }
1245
1246 error:
1247         health_code_update();
1248         return ret;
1249 }