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