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 error:
 172         free(pathname);
 173         return ret;
 174 }
 175
 176 /*
 177  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 178  */
 179 int kernel_consumer_add_metadata(struct consumer_socket *sock,
 180                 struct ltt_kernel_session *session, unsigned int monitor)
 181 {
 182         int ret;
 183         char *pathname;
 184         struct lttcomm_consumer_msg lkm;
 185         struct consumer_output *consumer;
 186
 187         /* Safety net */
 188         assert(session);
 189         assert(session->consumer);
 190         assert(sock);
 191
 192         DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);
 193
 194         /* Get consumer output pointer */
 195         consumer = session->consumer;
 196
 197         if (monitor) {
 198                 pathname = create_channel_path(consumer, session->uid, session->gid);
 199         } else {
 200                 /* Empty path. */
 201                 pathname = strdup("");
 202         }
 203         if (!pathname) {
 204                 ret = -1;
 205                 goto error;
 206         }
 207
 208         /* Prep channel message structure */
 209         consumer_init_channel_comm_msg(&lkm,
 210                         LTTNG_CONSUMER_ADD_CHANNEL,
 211                         session->metadata->fd,
 212                         session->id,
 213                         pathname,
 214                         session->uid,
 215                         session->gid,
 216                         consumer->net_seq_index,
 217                         DEFAULT_METADATA_NAME,
 218                         1,
 219                         DEFAULT_KERNEL_CHANNEL_OUTPUT,
 220                         CONSUMER_CHANNEL_TYPE_METADATA,
 221                         0, 0,
 222                         monitor, 0, 0);
 223
 224         health_code_update();
 225
 226         ret = consumer_send_channel(sock, &lkm);
 227         if (ret < 0) {
 228                 goto error;
 229         }
 230
 231         health_code_update();
 232
 233         /* Prep stream message structure */
 234         consumer_init_stream_comm_msg(&lkm,
 235                         LTTNG_CONSUMER_ADD_STREAM,
 236                         session->metadata->fd,
 237                         session->metadata_stream_fd,
 238                         0); /* CPU: 0 for metadata. */
 239
 240         health_code_update();
 241
 242         /* Send stream and file descriptor */
 243         ret = consumer_send_stream(sock, consumer, &lkm,
 244                         &session->metadata_stream_fd, 1);
 245         if (ret < 0) {
 246                 goto error;
 247         }
 248
 249         health_code_update();
 250
 251 error:
 252         free(pathname);
 253         return ret;
 254 }
 255
 256 /*
 257  * Sending a single stream to the consumer with command ADD_STREAM.
 258  */
 259 int kernel_consumer_add_stream(struct consumer_socket *sock,
 260                 struct ltt_kernel_channel *channel, struct ltt_kernel_stream *stream,
 261                 struct ltt_kernel_session *session, unsigned int monitor)
 262 {
 263         int ret;
 264         struct lttcomm_consumer_msg lkm;
 265         struct consumer_output *consumer;
 266
 267         assert(channel);
 268         assert(stream);
 269         assert(session);
 270         assert(session->consumer);
 271         assert(sock);
 272
 273         DBG("Sending stream %d of channel %s to kernel consumer",
 274                         stream->fd, channel->channel->name);
 275
 276         /* Get consumer output pointer */
 277         consumer = session->consumer;
 278
 279         /* Prep stream consumer message */
 280         consumer_init_stream_comm_msg(&lkm,
 281                         LTTNG_CONSUMER_ADD_STREAM,
 282                         channel->fd,
 283                         stream->fd,
 284                         stream->cpu);
 285
 286         health_code_update();
 287
 288         /* Send stream and file descriptor */
 289         ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
 290         if (ret < 0) {
 291                 goto error;
 292         }
 293
 294         health_code_update();
 295
 296 error:
 297         return ret;
 298 }
 299
 300 /*
 301  * Sending the notification that all streams were sent with STREAMS_SENT.
 302  */
 303 int kernel_consumer_streams_sent(struct consumer_socket *sock,
 304                 struct ltt_kernel_session *session, uint64_t channel_key)
 305 {
 306         int ret;
 307         struct lttcomm_consumer_msg lkm;
 308         struct consumer_output *consumer;
 309
 310         assert(sock);
 311         assert(session);
 312
 313         DBG("Sending streams_sent");
 314         /* Get consumer output pointer */
 315         consumer = session->consumer;
 316
 317         /* Prep stream consumer message */
 318         consumer_init_streams_sent_comm_msg(&lkm,
 319                         LTTNG_CONSUMER_STREAMS_SENT,
 320                         channel_key, consumer->net_seq_index);
 321
 322         health_code_update();
 323
 324         /* Send stream and file descriptor */
 325         ret = consumer_send_msg(sock, &lkm);
 326         if (ret < 0) {
 327                 goto error;
 328         }
 329
 330 error:
 331         return ret;
 332 }
 333
 334 /*
 335  * Send all stream fds of kernel channel to the consumer.
 336  */
 337 int kernel_consumer_send_channel_stream(struct consumer_socket *sock,
 338                 struct ltt_kernel_channel *channel, struct ltt_kernel_session *session,
 339                 unsigned int monitor)
 340 {
 341         int ret;
 342         struct ltt_kernel_stream *stream;
 343
 344         /* Safety net */
 345         assert(channel);
 346         assert(session);
 347         assert(session->consumer);
 348         assert(sock);
 349
 350         /* Bail out if consumer is disabled */
 351         if (!session->consumer->enabled) {
 352                 ret = LTTNG_OK;
 353                 goto error;
 354         }
 355
 356         DBG("Sending streams of channel %s to kernel consumer",
 357                         channel->channel->name);
 358
 359         ret = kernel_consumer_add_channel(sock, channel, session, monitor);
 360         if (ret < 0) {
 361                 goto error;
 362         }
 363
 364         /* Send streams */
 365         cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
 366                 if (!stream->fd || stream->sent_to_consumer) {
 367                         continue;
 368                 }
 369
 370                 /* Add stream on the kernel consumer side. */
 371                 ret = kernel_consumer_add_stream(sock, channel, stream, session,
 372                                 monitor);
 373                 if (ret < 0) {
 374                         goto error;
 375                 }
 376                 stream->sent_to_consumer = true;
 377         }
 378
 379 error:
 380         return ret;
 381 }
 382
 383 /*
 384  * Send all stream fds of the kernel session to the consumer.
 385  */
 386 int kernel_consumer_send_session(struct consumer_socket *sock,
 387                 struct ltt_kernel_session *session)
 388 {
 389         int ret, monitor = 0;
 390         struct ltt_kernel_channel *chan;
 391
 392         /* Safety net */
 393         assert(session);
 394         assert(session->consumer);
 395         assert(sock);
 396
 397         /* Bail out if consumer is disabled */
 398         if (!session->consumer->enabled) {
 399                 ret = LTTNG_OK;
 400                 goto error;
 401         }
 402
 403         /* Don't monitor the streams on the consumer if in flight recorder. */
 404         if (session->output_traces) {
 405                 monitor = 1;
 406         }
 407
 408         DBG("Sending session stream to kernel consumer");
 409
 410         if (session->metadata_stream_fd >= 0 && session->metadata) {
 411                 ret = kernel_consumer_add_metadata(sock, session, monitor);
 412                 if (ret < 0) {
 413                         goto error;
 414                 }
 415         }
 416
 417         /* Send channel and streams of it */
 418         cds_list_for_each_entry(chan, &session->channel_list.head, list) {
 419                 ret = kernel_consumer_send_channel_stream(sock, chan, session,
 420                                 monitor);
 421                 if (ret < 0) {
 422                         goto error;
 423                 }
 424                 if (monitor) {
 425                         /*
 426                          * Inform the relay that all the streams for the
 427                          * channel were sent.
 428                          */
 429                         ret = kernel_consumer_streams_sent(sock, session, chan->fd);
 430                         if (ret < 0) {
 431                                 goto error;
 432                         }
 433                 }
 434         }
 435
 436         DBG("Kernel consumer FDs of metadata and channel streams sent");
 437
 438         session->consumer_fds_sent = 1;
 439         return 0;
 440
 441 error:
 442         return ret;
 443 }
 444
 445 int kernel_consumer_destroy_channel(struct consumer_socket *socket,
 446                 struct ltt_kernel_channel *channel)
 447 {
 448         int ret;
 449         struct lttcomm_consumer_msg msg;
 450
 451         assert(channel);
 452         assert(socket);
 453
 454         DBG("Sending kernel consumer destroy channel key %d", channel->fd);
 455
 456         memset(&msg, 0, sizeof(msg));
 457         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 458         msg.u.destroy_channel.key = channel->fd;
 459
 460         pthread_mutex_lock(socket->lock);
 461         health_code_update();
 462
 463         ret = consumer_send_msg(socket, &msg);
 464         if (ret < 0) {
 465                 goto error;
 466         }
 467
 468 error:
 469         health_code_update();
 470         pthread_mutex_unlock(socket->lock);
 471         return ret;
 472 }
 473
 474 int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
 475                 struct ltt_kernel_metadata *metadata)
 476 {
 477         int ret;
 478         struct lttcomm_consumer_msg msg;
 479
 480         assert(metadata);
 481         assert(socket);
 482
 483         DBG("Sending kernel consumer destroy channel key %d", metadata->fd);
 484
 485         memset(&msg, 0, sizeof(msg));
 486         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 487         msg.u.destroy_channel.key = metadata->fd;
 488
 489         pthread_mutex_lock(socket->lock);
 490         health_code_update();
 491
 492         ret = consumer_send_msg(socket, &msg);
 493         if (ret < 0) {
 494                 goto error;
 495         }
 496
 497 error:
 498         health_code_update();
 499         pthread_mutex_unlock(socket->lock);
 500         return ret;
 501 }