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