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