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