src/bin/lttng-sessiond/kernel-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 _LGPL_SOURCE
  19 #include <stdio.h>
  20 #include <stdlib.h>
  21 #include <sys/stat.h>
  22 #include <unistd.h>
  23
  24 #include <common/common.h>
  25 #include <common/defaults.h>
  26 #include <common/compat/string.h>
  27
  28 #include "consumer.h"
  29 #include "health-sessiond.h"
  30 #include "kernel-consumer.h"
  31 #include "notification-thread-commands.h"
  32 #include "session.h"
  33 #include "lttng-sessiond.h"
  34
  35 static char *create_channel_path(struct consumer_output *consumer,
  36                 uid_t uid, gid_t gid)
  37 {
  38         int ret;
  39         char tmp_path[PATH_MAX];
  40         char *pathname = NULL;
  41
  42         assert(consumer);
  43
  44         /* Get the right path name destination */
  45         if (consumer->type == CONSUMER_DST_LOCAL) {
  46                 /* Set application path to the destination path */
  47                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s",
  48                                 consumer->dst.trace_path, consumer->subdir);
  49                 if (ret < 0) {
  50                         PERROR("snprintf kernel channel path");
  51                         goto error;
  52                 }
  53                 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
  54                 if (!pathname) {
  55                         PERROR("lttng_strndup");
  56                         goto error;
  57                 }
  58
  59                 /* Create directory */
  60                 ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, uid, gid);
  61                 if (ret < 0) {
  62                         if (errno != EEXIST) {
  63                                 ERR("Trace directory creation error");
  64                                 goto error;
  65                         }
  66                 }
  67                 DBG3("Kernel local consumer tracefile path: %s", pathname);
  68         } else {
  69                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
  70                 if (ret < 0) {
  71                         PERROR("snprintf kernel metadata path");
  72                         goto error;
  73                 }
  74                 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
  75                 if (!pathname) {
  76                         PERROR("lttng_strndup");
  77                         goto error;
  78                 }
  79                 DBG3("Kernel network consumer subdir path: %s", pathname);
  80         }
  81
  82         return pathname;
  83
  84 error:
  85         free(pathname);
  86         return NULL;
  87 }
  88
  89 /*
  90  * Sending a single channel to the consumer with command ADD_CHANNEL.
  91  */
  92 int kernel_consumer_add_channel(struct consumer_socket *sock,
  93                 struct ltt_kernel_channel *channel,
  94                 struct ltt_kernel_session *ksession,
  95                 unsigned int monitor)
  96 {
  97         int ret;
  98         char *pathname;
  99         struct lttcomm_consumer_msg lkm;
 100         struct consumer_output *consumer;
 101         enum lttng_error_code status;
 102         struct ltt_session *session;
 103         struct lttng_channel_extended *channel_attr_extended;
 104
 105         /* Safety net */
 106         assert(channel);
 107         assert(ksession);
 108         assert(ksession->consumer);
 109
 110         consumer = ksession->consumer;
 111         channel_attr_extended = (struct lttng_channel_extended *)
 112                         channel->channel->attr.extended.ptr;
 113
 114         DBG("Kernel consumer adding channel %s to kernel consumer",
 115                         channel->channel->name);
 116
 117         if (monitor) {
 118                 pathname = create_channel_path(consumer, ksession->uid,
 119                                 ksession->gid);
 120         } else {
 121                 /* Empty path. */
 122                 pathname = strdup("");
 123         }
 124         if (!pathname) {
 125                 ret = -1;
 126                 goto error;
 127         }
 128
 129         /* Prep channel message structure */
 130         consumer_init_channel_comm_msg(&lkm,
 131                         LTTNG_CONSUMER_ADD_CHANNEL,
 132                         channel->fd,
 133                         ksession->id,
 134                         pathname,
 135                         ksession->uid,
 136                         ksession->gid,
 137                         consumer->net_seq_index,
 138                         channel->channel->name,
 139                         channel->stream_count,
 140                         channel->channel->attr.output,
 141                         CONSUMER_CHANNEL_TYPE_DATA,
 142                         channel->channel->attr.tracefile_size,
 143                         channel->channel->attr.tracefile_count,
 144                         monitor,
 145                         channel->channel->attr.live_timer_interval,
 146                         channel_attr_extended->monitor_timer_interval);
 147
 148         health_code_update();
 149
 150         ret = consumer_send_channel(sock, &lkm);
 151         if (ret < 0) {
 152                 goto error;
 153         }
 154
 155         health_code_update();
 156         rcu_read_lock();
 157         session = session_find_by_id(ksession->id);
 158         assert(session);
 159
 160         status = notification_thread_command_add_channel(
 161                         notification_thread_handle, session->name,
 162                         ksession->uid, ksession->gid,
 163                         channel->channel->name, channel->fd,
 164                         LTTNG_DOMAIN_KERNEL,
 165                         channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
 166         rcu_read_unlock();
 167         if (status != LTTNG_OK) {
 168                 ret = -1;
 169                 goto error;
 170         }
 171
 172         channel->published_to_notification_thread = true;
 173
 174 error:
 175         free(pathname);
 176         return ret;
 177 }
 178
 179 /*
 180  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 181  *
 182  * The consumer socket lock must be held by the caller.
 183  */
 184 int kernel_consumer_add_metadata(struct consumer_socket *sock,
 185                 struct ltt_kernel_session *session, unsigned int monitor)
 186 {
 187         int ret;
 188         char *pathname;
 189         struct lttcomm_consumer_msg lkm;
 190         struct consumer_output *consumer;
 191
 192         /* Safety net */
 193         assert(session);
 194         assert(session->consumer);
 195         assert(sock);
 196
 197         DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);
 198
 199         /* Get consumer output pointer */
 200         consumer = session->consumer;
 201
 202         if (monitor) {
 203                 pathname = create_channel_path(consumer, session->uid, session->gid);
 204         } else {
 205                 /* Empty path. */
 206                 pathname = strdup("");
 207         }
 208         if (!pathname) {
 209                 ret = -1;
 210                 goto error;
 211         }
 212
 213         /* Prep channel message structure */
 214         consumer_init_channel_comm_msg(&lkm,
 215                         LTTNG_CONSUMER_ADD_CHANNEL,
 216                         session->metadata->fd,
 217                         session->id,
 218                         pathname,
 219                         session->uid,
 220                         session->gid,
 221                         consumer->net_seq_index,
 222                         DEFAULT_METADATA_NAME,
 223                         1,
 224                         DEFAULT_KERNEL_CHANNEL_OUTPUT,
 225                         CONSUMER_CHANNEL_TYPE_METADATA,
 226                         0, 0,
 227                         monitor, 0, 0);
 228
 229         health_code_update();
 230
 231         ret = consumer_send_channel(sock, &lkm);
 232         if (ret < 0) {
 233                 goto error;
 234         }
 235
 236         health_code_update();
 237
 238         /* Prep stream message structure */
 239         consumer_init_stream_comm_msg(&lkm,
 240                         LTTNG_CONSUMER_ADD_STREAM,
 241                         session->metadata->fd,
 242                         session->metadata_stream_fd,
 243                         0); /* CPU: 0 for metadata. */
 244
 245         health_code_update();
 246
 247         /* Send stream and file descriptor */
 248         ret = consumer_send_stream(sock, consumer, &lkm,
 249                         &session->metadata_stream_fd, 1);
 250         if (ret < 0) {
 251                 goto error;
 252         }
 253
 254         health_code_update();
 255
 256 error:
 257         free(pathname);
 258         return ret;
 259 }
 260
 261 /*
 262  * Sending a single stream to the consumer with command ADD_STREAM.
 263  */
 264 int kernel_consumer_add_stream(struct consumer_socket *sock,
 265                 struct ltt_kernel_channel *channel, struct ltt_kernel_stream *stream,
 266                 struct ltt_kernel_session *session, unsigned int monitor)
 267 {
 268         int ret;
 269         struct lttcomm_consumer_msg lkm;
 270         struct consumer_output *consumer;
 271
 272         assert(channel);
 273         assert(stream);
 274         assert(session);
 275         assert(session->consumer);
 276         assert(sock);
 277
 278         DBG("Sending stream %d of channel %s to kernel consumer",
 279                         stream->fd, channel->channel->name);
 280
 281         /* Get consumer output pointer */
 282         consumer = session->consumer;
 283
 284         /* Prep stream consumer message */
 285         consumer_init_stream_comm_msg(&lkm,
 286                         LTTNG_CONSUMER_ADD_STREAM,
 287                         channel->fd,
 288                         stream->fd,
 289                         stream->cpu);
 290
 291         health_code_update();
 292
 293         /* Send stream and file descriptor */
 294         ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
 295         if (ret < 0) {
 296                 goto error;
 297         }
 298
 299         health_code_update();
 300
 301 error:
 302         return ret;
 303 }
 304
 305 /*
 306  * Sending the notification that all streams were sent with STREAMS_SENT.
 307  */
 308 int kernel_consumer_streams_sent(struct consumer_socket *sock,
 309                 struct ltt_kernel_session *session, uint64_t channel_key)
 310 {
 311         int ret;
 312         struct lttcomm_consumer_msg lkm;
 313         struct consumer_output *consumer;
 314
 315         assert(sock);
 316         assert(session);
 317
 318         DBG("Sending streams_sent");
 319         /* Get consumer output pointer */
 320         consumer = session->consumer;
 321
 322         /* Prep stream consumer message */
 323         consumer_init_streams_sent_comm_msg(&lkm,
 324                         LTTNG_CONSUMER_STREAMS_SENT,
 325                         channel_key, consumer->net_seq_index);
 326
 327         health_code_update();
 328
 329         /* Send stream and file descriptor */
 330         ret = consumer_send_msg(sock, &lkm);
 331         if (ret < 0) {
 332                 goto error;
 333         }
 334
 335 error:
 336         return ret;
 337 }
 338
 339 /*
 340  * Send all stream fds of kernel channel to the consumer.
 341  *
 342  * The consumer socket lock must be held by the caller.
 343  */
 344 int kernel_consumer_send_channel_stream(struct consumer_socket *sock,
 345                 struct ltt_kernel_channel *channel, struct ltt_kernel_session *session,
 346                 unsigned int monitor)
 347 {
 348         int ret = LTTNG_OK;
 349         struct ltt_kernel_stream *stream;
 350
 351         /* Safety net */
 352         assert(channel);
 353         assert(session);
 354         assert(session->consumer);
 355         assert(sock);
 356
 357         /* Bail out if consumer is disabled */
 358         if (!session->consumer->enabled) {
 359                 ret = LTTNG_OK;
 360                 goto error;
 361         }
 362
 363         DBG("Sending streams of channel %s to kernel consumer",
 364                         channel->channel->name);
 365
 366         if (!channel->sent_to_consumer) {
 367                 ret = kernel_consumer_add_channel(sock, channel, session, monitor);
 368                 if (ret < 0) {
 369                         goto error;
 370                 }
 371                 channel->sent_to_consumer = true;
 372         }
 373
 374         /* Send streams */
 375         cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
 376                 if (!stream->fd || stream->sent_to_consumer) {
 377                         continue;
 378                 }
 379
 380                 /* Add stream on the kernel consumer side. */
 381                 ret = kernel_consumer_add_stream(sock, channel, stream, session,
 382                                 monitor);
 383                 if (ret < 0) {
 384                         goto error;
 385                 }
 386                 stream->sent_to_consumer = true;
 387         }
 388
 389 error:
 390         return ret;
 391 }
 392
 393 /*
 394  * Send all stream fds of the kernel session to the consumer.
 395  *
 396  * The consumer socket lock must be held by the caller.
 397  */
 398 int kernel_consumer_send_session(struct consumer_socket *sock,
 399                 struct ltt_kernel_session *session)
 400 {
 401         int ret, monitor = 0;
 402         struct ltt_kernel_channel *chan;
 403
 404         /* Safety net */
 405         assert(session);
 406         assert(session->consumer);
 407         assert(sock);
 408
 409         /* Bail out if consumer is disabled */
 410         if (!session->consumer->enabled) {
 411                 ret = LTTNG_OK;
 412                 goto error;
 413         }
 414
 415         /* Don't monitor the streams on the consumer if in flight recorder. */
 416         if (session->output_traces) {
 417                 monitor = 1;
 418         }
 419
 420         DBG("Sending session stream to kernel consumer");
 421
 422         if (session->metadata_stream_fd >= 0 && session->metadata) {
 423                 ret = kernel_consumer_add_metadata(sock, session, monitor);
 424                 if (ret < 0) {
 425                         goto error;
 426                 }
 427         }
 428
 429         /* Send channel and streams of it */
 430         cds_list_for_each_entry(chan, &session->channel_list.head, list) {
 431                 ret = kernel_consumer_send_channel_stream(sock, chan, session,
 432                                 monitor);
 433                 if (ret < 0) {
 434                         goto error;
 435                 }
 436                 if (monitor) {
 437                         /*
 438                          * Inform the relay that all the streams for the
 439                          * channel were sent.
 440                          */
 441                         ret = kernel_consumer_streams_sent(sock, session, chan->fd);
 442                         if (ret < 0) {
 443                                 goto error;
 444                         }
 445                 }
 446         }
 447
 448         DBG("Kernel consumer FDs of metadata and channel streams sent");
 449
 450         session->consumer_fds_sent = 1;
 451         return 0;
 452
 453 error:
 454         return ret;
 455 }
 456
 457 int kernel_consumer_destroy_channel(struct consumer_socket *socket,
 458                 struct ltt_kernel_channel *channel)
 459 {
 460         int ret;
 461         struct lttcomm_consumer_msg msg;
 462
 463         assert(channel);
 464         assert(socket);
 465
 466         DBG("Sending kernel consumer destroy channel key %d", channel->fd);
 467
 468         memset(&msg, 0, sizeof(msg));
 469         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 470         msg.u.destroy_channel.key = channel->fd;
 471
 472         pthread_mutex_lock(socket->lock);
 473         health_code_update();
 474
 475         ret = consumer_send_msg(socket, &msg);
 476         if (ret < 0) {
 477                 goto error;
 478         }
 479
 480 error:
 481         health_code_update();
 482         pthread_mutex_unlock(socket->lock);
 483         return ret;
 484 }
 485
 486 int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
 487                 struct ltt_kernel_metadata *metadata)
 488 {
 489         int ret;
 490         struct lttcomm_consumer_msg msg;
 491
 492         assert(metadata);
 493         assert(socket);
 494
 495         DBG("Sending kernel consumer destroy channel key %d", metadata->fd);
 496
 497         memset(&msg, 0, sizeof(msg));
 498         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 499         msg.u.destroy_channel.key = metadata->fd;
 500
 501         pthread_mutex_lock(socket->lock);
 502         health_code_update();
 503
 504         ret = consumer_send_msg(socket, &msg);
 505         if (ret < 0) {
 506                 goto error;
 507         }
 508
 509 error:
 510         health_code_update();
 511         pthread_mutex_unlock(socket->lock);
 512         return ret;
 513 }