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
  32 static char *create_channel_path(struct consumer_output *consumer,
  33                 uid_t uid, gid_t gid)
  34 {
  35         int ret;
  36         char tmp_path[PATH_MAX];
  37         char *pathname = NULL;
  38
  39         assert(consumer);
  40
  41         /* Get the right path name destination */
  42         if (consumer->type == CONSUMER_DST_LOCAL) {
  43                 /* Set application path to the destination path */
  44                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s",
  45                                 consumer->dst.trace_path, consumer->subdir);
  46                 if (ret < 0) {
  47                         PERROR("snprintf kernel channel path");
  48                         goto error;
  49                 }
  50                 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
  51                 if (!pathname) {
  52                         PERROR("lttng_strndup");
  53                         goto error;
  54                 }
  55
  56                 /* Create directory */
  57                 ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG, uid, gid);
  58                 if (ret < 0) {
  59                         if (errno != EEXIST) {
  60                                 ERR("Trace directory creation error");
  61                                 goto error;
  62                         }
  63                 }
  64                 DBG3("Kernel local consumer tracefile path: %s", pathname);
  65         } else {
  66                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s", consumer->subdir);
  67                 if (ret < 0) {
  68                         PERROR("snprintf kernel metadata path");
  69                         goto error;
  70                 }
  71                 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
  72                 if (!pathname) {
  73                         PERROR("lttng_strndup");
  74                         goto error;
  75                 }
  76                 DBG3("Kernel network consumer subdir path: %s", pathname);
  77         }
  78
  79         return pathname;
  80
  81 error:
  82         free(pathname);
  83         return NULL;
  84 }
  85
  86 /*
  87  * Sending a single channel to the consumer with command ADD_CHANNEL.
  88  */
  89 int kernel_consumer_add_channel(struct consumer_socket *sock,
  90                 struct ltt_kernel_channel *channel, struct ltt_kernel_session *session,
  91                 unsigned int monitor)
  92 {
  93         int ret;
  94         char *pathname;
  95         struct lttcomm_consumer_msg lkm;
  96         struct consumer_output *consumer;
  97
  98         /* Safety net */
  99         assert(channel);
 100         assert(session);
 101         assert(session->consumer);
 102
 103         consumer = session->consumer;
 104
 105         DBG("Kernel consumer adding channel %s to kernel consumer",
 106                         channel->channel->name);
 107
 108         if (monitor) {
 109                 pathname = create_channel_path(consumer, session->uid, session->gid);
 110         } else {
 111                 /* Empty path. */
 112                 pathname = strdup("");
 113         }
 114         if (!pathname) {
 115                 ret = -1;
 116                 goto error;
 117         }
 118
 119         /* Prep channel message structure */
 120         consumer_init_channel_comm_msg(&lkm,
 121                         LTTNG_CONSUMER_ADD_CHANNEL,
 122                         channel->fd,
 123                         session->id,
 124                         pathname,
 125                         session->uid,
 126                         session->gid,
 127                         consumer->net_seq_index,
 128                         channel->channel->name,
 129                         channel->stream_count,
 130                         channel->channel->attr.output,
 131                         CONSUMER_CHANNEL_TYPE_DATA,
 132                         channel->channel->attr.tracefile_size,
 133                         channel->channel->attr.tracefile_count,
 134                         monitor,
 135                         channel->channel->attr.live_timer_interval);
 136
 137         health_code_update();
 138
 139         ret = consumer_send_channel(sock, &lkm);
 140         if (ret < 0) {
 141                 goto error;
 142         }
 143
 144         health_code_update();
 145
 146 error:
 147         free(pathname);
 148         return ret;
 149 }
 150
 151 /*
 152  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 153  */
 154 int kernel_consumer_add_metadata(struct consumer_socket *sock,
 155                 struct ltt_kernel_session *session, unsigned int monitor)
 156 {
 157         int ret;
 158         char *pathname;
 159         struct lttcomm_consumer_msg lkm;
 160         struct consumer_output *consumer;
 161
 162         /* Safety net */
 163         assert(session);
 164         assert(session->consumer);
 165         assert(sock);
 166
 167         DBG("Sending metadata %d to kernel consumer", session->metadata_stream_fd);
 168
 169         /* Get consumer output pointer */
 170         consumer = session->consumer;
 171
 172         if (monitor) {
 173                 pathname = create_channel_path(consumer, session->uid, session->gid);
 174         } else {
 175                 /* Empty path. */
 176                 pathname = strdup("");
 177         }
 178         if (!pathname) {
 179                 ret = -1;
 180                 goto error;
 181         }
 182
 183         /* Prep channel message structure */
 184         consumer_init_channel_comm_msg(&lkm,
 185                         LTTNG_CONSUMER_ADD_CHANNEL,
 186                         session->metadata->fd,
 187                         session->id,
 188                         pathname,
 189                         session->uid,
 190                         session->gid,
 191                         consumer->net_seq_index,
 192                         DEFAULT_METADATA_NAME,
 193                         1,
 194                         DEFAULT_KERNEL_CHANNEL_OUTPUT,
 195                         CONSUMER_CHANNEL_TYPE_METADATA,
 196                         0, 0,
 197                         monitor, 0);
 198
 199         health_code_update();
 200
 201         ret = consumer_send_channel(sock, &lkm);
 202         if (ret < 0) {
 203                 goto error;
 204         }
 205
 206         health_code_update();
 207
 208         /* Prep stream message structure */
 209         consumer_init_stream_comm_msg(&lkm,
 210                         LTTNG_CONSUMER_ADD_STREAM,
 211                         session->metadata->fd,
 212                         session->metadata_stream_fd,
 213                         0); /* CPU: 0 for metadata. */
 214
 215         health_code_update();
 216
 217         /* Send stream and file descriptor */
 218         ret = consumer_send_stream(sock, consumer, &lkm,
 219                         &session->metadata_stream_fd, 1);
 220         if (ret < 0) {
 221                 goto error;
 222         }
 223
 224         health_code_update();
 225
 226 error:
 227         free(pathname);
 228         return ret;
 229 }
 230
 231 /*
 232  * Sending a single stream to the consumer with command ADD_STREAM.
 233  */
 234 int kernel_consumer_add_stream(struct consumer_socket *sock,
 235                 struct ltt_kernel_channel *channel, struct ltt_kernel_stream *stream,
 236                 struct ltt_kernel_session *session, unsigned int monitor)
 237 {
 238         int ret;
 239         struct lttcomm_consumer_msg lkm;
 240         struct consumer_output *consumer;
 241
 242         assert(channel);
 243         assert(stream);
 244         assert(session);
 245         assert(session->consumer);
 246         assert(sock);
 247
 248         DBG("Sending stream %d of channel %s to kernel consumer",
 249                         stream->fd, channel->channel->name);
 250
 251         /* Get consumer output pointer */
 252         consumer = session->consumer;
 253
 254         /* Prep stream consumer message */
 255         consumer_init_stream_comm_msg(&lkm,
 256                         LTTNG_CONSUMER_ADD_STREAM,
 257                         channel->fd,
 258                         stream->fd,
 259                         stream->cpu);
 260
 261         health_code_update();
 262
 263         /* Send stream and file descriptor */
 264         ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
 265         if (ret < 0) {
 266                 goto error;
 267         }
 268
 269         health_code_update();
 270
 271 error:
 272         return ret;
 273 }
 274
 275 /*
 276  * Sending the notification that all streams were sent with STREAMS_SENT.
 277  */
 278 int kernel_consumer_streams_sent(struct consumer_socket *sock,
 279                 struct ltt_kernel_session *session, uint64_t channel_key)
 280 {
 281         int ret;
 282         struct lttcomm_consumer_msg lkm;
 283         struct consumer_output *consumer;
 284
 285         assert(sock);
 286         assert(session);
 287
 288         DBG("Sending streams_sent");
 289         /* Get consumer output pointer */
 290         consumer = session->consumer;
 291
 292         /* Prep stream consumer message */
 293         consumer_init_streams_sent_comm_msg(&lkm,
 294                         LTTNG_CONSUMER_STREAMS_SENT,
 295                         channel_key, consumer->net_seq_index);
 296
 297         health_code_update();
 298
 299         /* Send stream and file descriptor */
 300         ret = consumer_send_msg(sock, &lkm);
 301         if (ret < 0) {
 302                 goto error;
 303         }
 304
 305 error:
 306         return ret;
 307 }
 308
 309 /*
 310  * Send all stream fds of kernel channel to the consumer.
 311  */
 312 int kernel_consumer_send_channel_stream(struct consumer_socket *sock,
 313                 struct ltt_kernel_channel *channel, struct ltt_kernel_session *session,
 314                 unsigned int monitor)
 315 {
 316         int ret;
 317         struct ltt_kernel_stream *stream;
 318
 319         /* Safety net */
 320         assert(channel);
 321         assert(session);
 322         assert(session->consumer);
 323         assert(sock);
 324
 325         /* Bail out if consumer is disabled */
 326         if (!session->consumer->enabled) {
 327                 ret = LTTNG_OK;
 328                 goto error;
 329         }
 330
 331         DBG("Sending streams of channel %s to kernel consumer",
 332                         channel->channel->name);
 333
 334         ret = kernel_consumer_add_channel(sock, channel, session, monitor);
 335         if (ret < 0) {
 336                 goto error;
 337         }
 338
 339         /* Send streams */
 340         cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
 341                 if (!stream->fd || stream->sent_to_consumer) {
 342                         continue;
 343                 }
 344
 345                 /* Add stream on the kernel consumer side. */
 346                 ret = kernel_consumer_add_stream(sock, channel, stream, session,
 347                                 monitor);
 348                 if (ret < 0) {
 349                         goto error;
 350                 }
 351                 stream->sent_to_consumer = true;
 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 }