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                 uint64_t session_id_per_pid,
 726                 unsigned int monitor)
 727 {
 728         assert(msg);
 729
 730         /* Zeroed structure */
 731         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 732
 733         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 734         msg->u.ask_channel.subbuf_size = subbuf_size;
 735         msg->u.ask_channel.num_subbuf = num_subbuf ;
 736         msg->u.ask_channel.overwrite = overwrite;
 737         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 738         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 739         msg->u.ask_channel.output = output;
 740         msg->u.ask_channel.type = type;
 741         msg->u.ask_channel.session_id = session_id;
 742         msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
 743         msg->u.ask_channel.uid = uid;
 744         msg->u.ask_channel.gid = gid;
 745         msg->u.ask_channel.relayd_id = relayd_id;
 746         msg->u.ask_channel.key = key;
 747         msg->u.ask_channel.chan_id = chan_id;
 748         msg->u.ask_channel.tracefile_size = tracefile_size;
 749         msg->u.ask_channel.tracefile_count = tracefile_count;
 750         msg->u.ask_channel.monitor = monitor;
 751
 752         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 753
 754         if (pathname) {
 755                 strncpy(msg->u.ask_channel.pathname, pathname,
 756                                 sizeof(msg->u.ask_channel.pathname));
 757                 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 758         }
 759
 760         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 761         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 762 }
 763
 764 /*
 765  * Init channel communication message structure.
 766  */
 767 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 768                 enum lttng_consumer_command cmd,
 769                 uint64_t channel_key,
 770                 uint64_t session_id,
 771                 const char *pathname,
 772                 uid_t uid,
 773                 gid_t gid,
 774                 uint64_t relayd_id,
 775                 const char *name,
 776                 unsigned int nb_init_streams,
 777                 enum lttng_event_output output,
 778                 int type,
 779                 uint64_t tracefile_size,
 780                 uint64_t tracefile_count,
 781                 unsigned int monitor)
 782 {
 783         assert(msg);
 784
 785         /* Zeroed structure */
 786         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 787
 788         /* Send channel */
 789         msg->cmd_type = cmd;
 790         msg->u.channel.channel_key = channel_key;
 791         msg->u.channel.session_id = session_id;
 792         msg->u.channel.uid = uid;
 793         msg->u.channel.gid = gid;
 794         msg->u.channel.relayd_id = relayd_id;
 795         msg->u.channel.nb_init_streams = nb_init_streams;
 796         msg->u.channel.output = output;
 797         msg->u.channel.type = type;
 798         msg->u.channel.tracefile_size = tracefile_size;
 799         msg->u.channel.tracefile_count = tracefile_count;
 800         msg->u.channel.monitor = monitor;
 801
 802         strncpy(msg->u.channel.pathname, pathname,
 803                         sizeof(msg->u.channel.pathname));
 804         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
 805
 806         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
 807         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 808 }
 809
 810 /*
 811  * Init stream communication message structure.
 812  */
 813 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 814                 enum lttng_consumer_command cmd,
 815                 uint64_t channel_key,
 816                 uint64_t stream_key,
 817                 int cpu)
 818 {
 819         assert(msg);
 820
 821         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 822
 823         msg->cmd_type = cmd;
 824         msg->u.stream.channel_key = channel_key;
 825         msg->u.stream.stream_key = stream_key;
 826         msg->u.stream.cpu = cpu;
 827 }
 828
 829 /*
 830  * Send stream communication structure to the consumer.
 831  */
 832 int consumer_send_stream(struct consumer_socket *sock,
 833                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
 834                 int *fds, size_t nb_fd)
 835 {
 836         int ret;
 837
 838         assert(msg);
 839         assert(dst);
 840         assert(sock);
 841         assert(fds);
 842
 843         /* Send on socket */
 844         ret = lttcomm_send_unix_sock(sock->fd, msg,
 845                         sizeof(struct lttcomm_consumer_msg));
 846         if (ret < 0) {
 847                 /* The above call will print a PERROR on error. */
 848                 DBG("Error when sending consumer stream on sock %d", sock->fd);
 849                 goto error;
 850         }
 851
 852         ret = consumer_recv_status_reply(sock);
 853         if (ret < 0) {
 854                 goto error;
 855         }
 856
 857         ret = consumer_send_fds(sock, fds, nb_fd);
 858         if (ret < 0) {
 859                 goto error;
 860         }
 861
 862 error:
 863         return ret;
 864 }
 865
 866 /*
 867  * Send relayd socket to consumer associated with a session name.
 868  *
 869  * On success return positive value. On error, negative value.
 870  */
 871 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 872                 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
 873                 enum lttng_stream_type type, uint64_t session_id)
 874 {
 875         int ret;
 876         struct lttcomm_consumer_msg msg;
 877
 878         /* Code flow error. Safety net. */
 879         assert(rsock);
 880         assert(consumer);
 881         assert(consumer_sock);
 882
 883         /* Bail out if consumer is disabled */
 884         if (!consumer->enabled) {
 885                 ret = LTTNG_OK;
 886                 goto error;
 887         }
 888
 889         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 890         /*
 891          * Assign network consumer output index using the temporary consumer since
 892          * this call should only be made from within a set_consumer_uri() function
 893          * call in the session daemon.
 894          */
 895         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 896         msg.u.relayd_sock.type = type;
 897         msg.u.relayd_sock.session_id = session_id;
 898         memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
 899
 900         DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
 901         ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
 902         if (ret < 0) {
 903                 /* The above call will print a PERROR on error. */
 904                 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
 905                 goto error;
 906         }
 907
 908         ret = consumer_recv_status_reply(consumer_sock);
 909         if (ret < 0) {
 910                 goto error;
 911         }
 912
 913         DBG3("Sending relayd socket file descriptor to consumer");
 914         ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
 915         if (ret < 0) {
 916                 goto error;
 917         }
 918
 919         DBG2("Consumer relayd socket sent");
 920
 921 error:
 922         return ret;
 923 }
 924
 925 /*
 926  * Set consumer subdirectory using the session name and a generated datetime if
 927  * needed. This is appended to the current subdirectory.
 928  */
 929 int consumer_set_subdir(struct consumer_output *consumer,
 930                 const char *session_name)
 931 {
 932         int ret = 0;
 933         unsigned int have_default_name = 0;
 934         char datetime[16], tmp_path[PATH_MAX];
 935         time_t rawtime;
 936         struct tm *timeinfo;
 937
 938         assert(consumer);
 939         assert(session_name);
 940
 941         memset(tmp_path, 0, sizeof(tmp_path));
 942
 943         /* Flag if we have a default session. */
 944         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
 945                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
 946                 have_default_name = 1;
 947         } else {
 948                 /* Get date and time for session path */
 949                 time(&rawtime);
 950                 timeinfo = localtime(&rawtime);
 951                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
 952         }
 953
 954         if (have_default_name) {
 955                 ret = snprintf(tmp_path, sizeof(tmp_path),
 956                                 "%s/%s", consumer->subdir, session_name);
 957         } else {
 958                 ret = snprintf(tmp_path, sizeof(tmp_path),
 959                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
 960         }
 961         if (ret < 0) {
 962                 PERROR("snprintf session name date");
 963                 goto error;
 964         }
 965
 966         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
 967         DBG2("Consumer subdir set to %s", consumer->subdir);
 968
 969 error:
 970         return ret;
 971 }
 972
 973 /*
 974  * Ask the consumer if the data is ready to read (NOT pending) for the specific
 975  * session id.
 976  *
 977  * This function has a different behavior with the consumer i.e. that it waits
 978  * for a reply from the consumer if yes or no the data is pending.
 979  */
 980 int consumer_is_data_pending(uint64_t session_id,
 981                 struct consumer_output *consumer)
 982 {
 983         int ret;
 984         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
 985         struct consumer_socket *socket;
 986         struct lttng_ht_iter iter;
 987         struct lttcomm_consumer_msg msg;
 988
 989         assert(consumer);
 990
 991         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
 992
 993         msg.u.data_pending.session_id = session_id;
 994
 995         DBG3("Consumer data pending for id %" PRIu64, session_id);
 996
 997         /* Send command for each consumer */
 998         rcu_read_lock();
 999         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1000                         node.node) {
1001                 /* Code flow error */
1002                 assert(socket->fd >= 0);
1003
1004                 pthread_mutex_lock(socket->lock);
1005
1006                 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
1007                 if (ret < 0) {
1008                         /* The above call will print a PERROR on error. */
1009                         DBG("Error on consumer is data pending on sock %d", socket->fd);
1010                         pthread_mutex_unlock(socket->lock);
1011                         goto error_unlock;
1012                 }
1013
1014                 /*
1015                  * No need for a recv reply status because the answer to the command is
1016                  * the reply status message.
1017                  */
1018
1019                 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
1020                 if (ret <= 0) {
1021                         if (ret == 0) {
1022                                 /* Orderly shutdown. Don't return 0 which means success. */
1023                                 ret = -1;
1024                         }
1025                         /* The above call will print a PERROR on error. */
1026                         DBG("Error on recv consumer is data pending on sock %d", socket->fd);
1027                         pthread_mutex_unlock(socket->lock);
1028                         goto error_unlock;
1029                 }
1030
1031                 pthread_mutex_unlock(socket->lock);
1032
1033                 if (ret_code == 1) {
1034                         break;
1035                 }
1036         }
1037         rcu_read_unlock();
1038
1039         DBG("Consumer data is %s pending for session id %" PRIu64,
1040                         ret_code == 1 ? "" : "NOT", session_id);
1041         return ret_code;
1042
1043 error_unlock:
1044         rcu_read_unlock();
1045         return -1;
1046 }
1047
1048 /*
1049  * Send a flush command to consumer using the given channel key.
1050  *
1051  * Return 0 on success else a negative value.
1052  */
1053 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1054 {
1055         int ret;
1056         struct lttcomm_consumer_msg msg;
1057
1058         assert(socket);
1059         assert(socket->fd >= 0);
1060
1061         DBG2("Consumer flush channel key %" PRIu64, key);
1062
1063         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1064         msg.u.flush_channel.key = key;
1065
1066         pthread_mutex_lock(socket->lock);
1067         health_code_update();
1068
1069         ret = consumer_send_msg(socket, &msg);
1070         if (ret < 0) {
1071                 goto end;
1072         }
1073
1074 end:
1075         health_code_update();
1076         pthread_mutex_unlock(socket->lock);
1077         return ret;
1078 }
1079
1080 /*
1081  * Send a close metdata command to consumer using the given channel key.
1082  *
1083  * Return 0 on success else a negative value.
1084  */
1085 int consumer_close_metadata(struct consumer_socket *socket,
1086                 uint64_t metadata_key)
1087 {
1088         int ret;
1089         struct lttcomm_consumer_msg msg;
1090
1091         assert(socket);
1092         assert(socket->fd >= 0);
1093
1094         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1095
1096         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1097         msg.u.close_metadata.key = metadata_key;
1098
1099         pthread_mutex_lock(socket->lock);
1100         health_code_update();
1101
1102         ret = consumer_send_msg(socket, &msg);
1103         if (ret < 0) {
1104                 goto end;
1105         }
1106
1107 end:
1108         health_code_update();
1109         pthread_mutex_unlock(socket->lock);
1110         return ret;
1111 }
1112
1113 /*
1114  * Send a setup metdata command to consumer using the given channel key.
1115  *
1116  * Return 0 on success else a negative value.
1117  */
1118 int consumer_setup_metadata(struct consumer_socket *socket,
1119                 uint64_t metadata_key)
1120 {
1121         int ret;
1122         struct lttcomm_consumer_msg msg;
1123
1124         assert(socket);
1125         assert(socket->fd >= 0);
1126
1127         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1128
1129         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1130         msg.u.setup_metadata.key = metadata_key;
1131
1132         pthread_mutex_lock(socket->lock);
1133         health_code_update();
1134
1135         ret = consumer_send_msg(socket, &msg);
1136         if (ret < 0) {
1137                 goto end;
1138         }
1139
1140 end:
1141         health_code_update();
1142         pthread_mutex_unlock(socket->lock);
1143         return ret;
1144 }
1145
1146 /*
1147  * Send metadata string to consumer. Socket lock MUST be acquired.
1148  *
1149  * Return 0 on success else a negative value.
1150  */
1151 int consumer_push_metadata(struct consumer_socket *socket,
1152                 uint64_t metadata_key, char *metadata_str, size_t len,
1153                 size_t target_offset)
1154 {
1155         int ret;
1156         struct lttcomm_consumer_msg msg;
1157
1158         assert(socket);
1159         assert(socket->fd >= 0);
1160
1161         DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1162
1163         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1164         msg.u.push_metadata.key = metadata_key;
1165         msg.u.push_metadata.target_offset = target_offset;
1166         msg.u.push_metadata.len = len;
1167
1168         health_code_update();
1169         ret = consumer_send_msg(socket, &msg);
1170         if (ret < 0 || len == 0) {
1171                 goto end;
1172         }
1173
1174         DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
1175
1176         ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1177         if (ret < 0) {
1178                 goto end;
1179         }
1180
1181         health_code_update();
1182         ret = consumer_recv_status_reply(socket);
1183         if (ret < 0) {
1184                 goto end;
1185         }
1186
1187 end:
1188         health_code_update();
1189         return ret;
1190 }
1191
1192 /*
1193  * Ask the consumer to snapshot a specific channel using the key.
1194  *
1195  * Return 0 on success or else a negative error.
1196  */
1197 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1198                 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1199                 const char *session_path, int wait)
1200 {
1201         int ret;
1202         struct lttcomm_consumer_msg msg;
1203
1204         assert(socket);
1205         assert(socket->fd >= 0);
1206         assert(output);
1207         assert(output->consumer);
1208
1209         DBG("Consumer snapshot channel key %" PRIu64, key);
1210
1211         memset(&msg, 0, sizeof(msg));
1212         msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1213         msg.u.snapshot_channel.key = key;
1214         msg.u.snapshot_channel.max_size = output->max_size;
1215         msg.u.snapshot_channel.metadata = metadata;
1216
1217         if (output->consumer->type == CONSUMER_DST_NET) {
1218                 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1219                 msg.u.snapshot_channel.use_relayd = 1;
1220                 ret = snprintf(msg.u.snapshot_channel.pathname,
1221                                 sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s",
1222                                 output->consumer->subdir, output->name, output->datetime,
1223                                 session_path);
1224                 if (ret < 0) {
1225                         ret = -LTTNG_ERR_NOMEM;
1226                         goto error;
1227                 }
1228         } else {
1229                 ret = snprintf(msg.u.snapshot_channel.pathname,
1230                                 sizeof(msg.u.snapshot_channel.pathname), "%s/%s-%s%s",
1231                                 output->consumer->dst.trace_path, output->name,
1232                                 output->datetime, session_path);
1233                 if (ret < 0) {
1234                         ret = -LTTNG_ERR_NOMEM;
1235                         goto error;
1236                 }
1237                 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1238
1239                 /* Create directory. Ignore if exist. */
1240                 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1241                                 S_IRWXU | S_IRWXG, uid, gid);
1242                 if (ret < 0) {
1243                         if (ret != -EEXIST) {
1244                                 ERR("Trace directory creation error");
1245                                 goto error;
1246                         }
1247                 }
1248         }
1249
1250         health_code_update();
1251         ret = consumer_send_msg(socket, &msg);
1252         if (ret < 0) {
1253                 goto error;
1254         }
1255
1256 error:
1257         health_code_update();
1258         return ret;
1259 }