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