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