src/bin/lttng-sessiond/ust-consumer.c

   1 /*
   2  * Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
   3  *
   4  * This program is free software; you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License, version 2 only,
   6  * as published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11  * GNU General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along
  14  * with this program; if not, write to the Free Software Foundation, Inc.,
  15  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _LGPL_SOURCE
  19 #include <errno.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <unistd.h>
  24 #include <inttypes.h>
  25
  26 #include <common/common.h>
  27 #include <common/consumer/consumer.h>
  28 #include <common/defaults.h>
  29
  30 #include "consumer.h"
  31 #include "health-sessiond.h"
  32 #include "ust-consumer.h"
  33 #include "buffer-registry.h"
  34 #include "session.h"
  35 #include "lttng-sessiond.h"
  36
  37 /*
  38  * Return allocated full pathname of the session using the consumer trace path
  39  * and subdir if available.
  40  *
  41  * The caller can safely free(3) the returned value. On error, NULL is
  42  * returned.
  43  */
  44 static char *setup_channel_trace_path(struct consumer_output *consumer,
  45                 struct ust_app_session *ua_sess)
  46 {
  47         int ret;
  48         char *pathname;
  49
  50         assert(consumer);
  51         assert(ua_sess);
  52
  53         health_code_update();
  54
  55         /*
  56          * Allocate the string ourself to make sure we never exceed
  57          * LTTNG_PATH_MAX.
  58          */
  59         pathname = zmalloc(LTTNG_PATH_MAX);
  60         if (!pathname) {
  61                 goto error;
  62         }
  63
  64         /* Get correct path name destination */
  65         if (consumer->type == CONSUMER_DST_NET &&
  66                         consumer->relay_major_version == 2 &&
  67                         consumer->relay_minor_version < 11) {
  68                 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s/%s%s",
  69                                 consumer->dst.net.base_dir,
  70                                 consumer->chunk_path, consumer->domain_subdir,
  71                                 ua_sess->path);
  72         } else {
  73                 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s",
  74                                 consumer->domain_subdir, ua_sess->path);
  75         }
  76         DBG3("Userspace consumer trace path relative to current trace chunk: \"%s\"",
  77                         pathname);
  78         if (ret < 0) {
  79                 PERROR("Failed to format channel path");
  80                 goto error;
  81         } else if (ret >= LTTNG_PATH_MAX) {
  82                 ERR("Truncation occurred while formatting channel path");
  83                 ret = -1;
  84                 goto error;
  85         }
  86
  87         return pathname;
  88 error:
  89         free(pathname);
  90         return NULL;
  91 }
  92
  93 /*
  94  * Send a single channel to the consumer using command ASK_CHANNEL_CREATION.
  95  *
  96  * Consumer socket lock MUST be acquired before calling this.
  97  */
  98 static int ask_channel_creation(struct ust_app_session *ua_sess,
  99                 struct ust_app_channel *ua_chan,
 100                 struct consumer_output *consumer,
 101                 struct consumer_socket *socket,
 102                 struct ust_registry_session *registry,
 103                 struct lttng_trace_chunk *trace_chunk)
 104 {
 105         int ret, output;
 106         uint32_t chan_id;
 107         uint64_t key, chan_reg_key;
 108         char *pathname = NULL;
 109         struct lttcomm_consumer_msg msg;
 110         struct ust_registry_channel *chan_reg;
 111         char shm_path[PATH_MAX] = "";
 112         char root_shm_path[PATH_MAX] = "";
 113         bool is_local_trace;
 114
 115         assert(ua_sess);
 116         assert(ua_chan);
 117         assert(socket);
 118         assert(consumer);
 119         assert(registry);
 120
 121         DBG2("Asking UST consumer for channel");
 122
 123         is_local_trace = consumer->net_seq_index == -1ULL;
 124         /* Format the channel's path (relative to the current trace chunk). */
 125         pathname = setup_channel_trace_path(consumer, ua_sess);
 126         if (!pathname) {
 127                 ret = -1;
 128                 goto error;
 129         }
 130
 131         if (is_local_trace && trace_chunk) {
 132                 enum lttng_trace_chunk_status chunk_status;
 133                 char *pathname_index;
 134
 135                 ret = asprintf(&pathname_index, "%s/" DEFAULT_INDEX_DIR,
 136                                 pathname);
 137                 if (ret < 0) {
 138                         ERR("Failed to format channel index directory");
 139                         ret = -1;
 140                         goto error;
 141                 }
 142
 143                 /*
 144                  * Create the index subdirectory which will take care
 145                  * of implicitly creating the channel's path.
 146                  */
 147                 chunk_status = lttng_trace_chunk_create_subdirectory(
 148                                 trace_chunk, pathname_index);
 149                 free(pathname_index);
 150                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 151                         ret = -1;
 152                         goto error;
 153                 }
 154         }
 155
 156         /* Depending on the buffer type, a different channel key is used. */
 157         if (ua_sess->buffer_type == LTTNG_BUFFER_PER_UID) {
 158                 chan_reg_key = ua_chan->tracing_channel_id;
 159         } else {
 160                 chan_reg_key = ua_chan->key;
 161         }
 162
 163         if (ua_chan->attr.type == LTTNG_UST_CHAN_METADATA) {
 164                 chan_id = -1U;
 165                 /*
 166                  * Metadata channels shm_path (buffers) are handled within
 167                  * session daemon. Consumer daemon should not try to create
 168                  * those buffer files.
 169                  */
 170         } else {
 171                 chan_reg = ust_registry_channel_find(registry, chan_reg_key);
 172                 assert(chan_reg);
 173                 chan_id = chan_reg->chan_id;
 174                 if (ua_sess->shm_path[0]) {
 175                         strncpy(shm_path, ua_sess->shm_path, sizeof(shm_path));
 176                         shm_path[sizeof(shm_path) - 1] = '\0';
 177                         strncat(shm_path, "/",
 178                                 sizeof(shm_path) - strlen(shm_path) - 1);
 179                         strncat(shm_path, ua_chan->name,
 180                                         sizeof(shm_path) - strlen(shm_path) - 1);
 181                                 strncat(shm_path, "_",
 182                                         sizeof(shm_path) - strlen(shm_path) - 1);
 183                 }
 184                 strncpy(root_shm_path, ua_sess->root_shm_path, sizeof(root_shm_path));
 185                 root_shm_path[sizeof(root_shm_path) - 1] = '\0';
 186         }
 187
 188         switch (ua_chan->attr.output) {
 189         case LTTNG_UST_MMAP:
 190         default:
 191                 output = LTTNG_EVENT_MMAP;
 192                 break;
 193         }
 194
 195         consumer_init_ask_channel_comm_msg(&msg,
 196                         ua_chan->attr.subbuf_size,
 197                         ua_chan->attr.num_subbuf,
 198                         ua_chan->attr.overwrite,
 199                         ua_chan->attr.switch_timer_interval,
 200                         ua_chan->attr.read_timer_interval,
 201                         ua_sess->live_timer_interval,
 202                         ua_chan->monitor_timer_interval,
 203                         output,
 204                         (int) ua_chan->attr.type,
 205                         ua_sess->tracing_id,
 206                         pathname,
 207                         ua_chan->name,
 208                         consumer->net_seq_index,
 209                         ua_chan->key,
 210                         registry->uuid,
 211                         chan_id,
 212                         ua_chan->tracefile_size,
 213                         ua_chan->tracefile_count,
 214                         ua_sess->id,
 215                         ua_sess->output_traces,
 216                         ua_sess->real_credentials.uid,
 217                         ua_chan->attr.blocking_timeout,
 218                         root_shm_path, shm_path,
 219                         trace_chunk,
 220                         &ua_sess->effective_credentials);
 221
 222         health_code_update();
 223
 224         ret = consumer_socket_send(socket, &msg, sizeof(msg));
 225         if (ret < 0) {
 226                 goto error;
 227         }
 228
 229         ret = consumer_recv_status_channel(socket, &key,
 230                         &ua_chan->expected_stream_count);
 231         if (ret < 0) {
 232                 goto error;
 233         }
 234         /* Communication protocol error. */
 235         assert(key == ua_chan->key);
 236         /* We need at least one where 1 stream for 1 cpu. */
 237         if (ua_sess->output_traces) {
 238                 assert(ua_chan->expected_stream_count > 0);
 239         }
 240
 241         DBG2("UST ask channel %" PRIu64 " successfully done with %u stream(s)", key,
 242                         ua_chan->expected_stream_count);
 243
 244 error:
 245         free(pathname);
 246         health_code_update();
 247         return ret;
 248 }
 249
 250 /*
 251  * Ask consumer to create a channel for a given session.
 252  *
 253  * Session list and rcu read side locks must be held by the caller.
 254  *
 255  * Returns 0 on success else a negative value.
 256  */
 257 int ust_consumer_ask_channel(struct ust_app_session *ua_sess,
 258                 struct ust_app_channel *ua_chan,
 259                 struct consumer_output *consumer,
 260                 struct consumer_socket *socket,
 261                 struct ust_registry_session *registry,
 262                 struct lttng_trace_chunk * trace_chunk)
 263 {
 264         int ret;
 265
 266         assert(ua_sess);
 267         assert(ua_chan);
 268         assert(consumer);
 269         assert(socket);
 270         assert(registry);
 271
 272         if (!consumer->enabled) {
 273                 ret = -LTTNG_ERR_NO_CONSUMER;
 274                 DBG3("Consumer is disabled");
 275                 goto error;
 276         }
 277
 278         pthread_mutex_lock(socket->lock);
 279         ret = ask_channel_creation(ua_sess, ua_chan, consumer, socket, registry,
 280                         trace_chunk);
 281         pthread_mutex_unlock(socket->lock);
 282         if (ret < 0) {
 283                 ERR("ask_channel_creation consumer command failed");
 284                 goto error;
 285         }
 286
 287 error:
 288         return ret;
 289 }
 290
 291 /*
 292  * Send a get channel command to consumer using the given channel key.  The
 293  * channel object is populated and the stream list.
 294  *
 295  * Return 0 on success else a negative value.
 296  */
 297 int ust_consumer_get_channel(struct consumer_socket *socket,
 298                 struct ust_app_channel *ua_chan)
 299 {
 300         int ret;
 301         struct lttcomm_consumer_msg msg;
 302
 303         assert(ua_chan);
 304         assert(socket);
 305
 306         memset(&msg, 0, sizeof(msg));
 307         msg.cmd_type = LTTNG_CONSUMER_GET_CHANNEL;
 308         msg.u.get_channel.key = ua_chan->key;
 309
 310         pthread_mutex_lock(socket->lock);
 311         health_code_update();
 312
 313         /* Send command and wait for OK reply. */
 314         ret = consumer_send_msg(socket, &msg);
 315         if (ret < 0) {
 316                 goto error;
 317         }
 318
 319         /* First, get the channel from consumer. */
 320         ret = ustctl_recv_channel_from_consumer(*socket->fd_ptr, &ua_chan->obj);
 321         if (ret < 0) {
 322                 if (ret != -EPIPE) {
 323                         ERR("Error recv channel from consumer %d with ret %d",
 324                                         *socket->fd_ptr, ret);
 325                 } else {
 326                         DBG3("UST app recv channel from consumer. Consumer is dead.");
 327                 }
 328                 goto error;
 329         }
 330
 331         /* Next, get all streams. */
 332         while (1) {
 333                 struct ust_app_stream *stream;
 334
 335                 /* Create UST stream */
 336                 stream = ust_app_alloc_stream();
 337                 if (stream == NULL) {
 338                         ret = -ENOMEM;
 339                         goto error;
 340                 }
 341
 342                 /* Stream object is populated by this call if successful. */
 343                 ret = ustctl_recv_stream_from_consumer(*socket->fd_ptr, &stream->obj);
 344                 if (ret < 0) {
 345                         free(stream);
 346                         if (ret == -LTTNG_UST_ERR_NOENT) {
 347                                 DBG3("UST app consumer has no more stream available");
 348                                 break;
 349                         }
 350                         if (ret != -EPIPE) {
 351                                 ERR("Recv stream from consumer %d with ret %d",
 352                                                 *socket->fd_ptr, ret);
 353                         } else {
 354                                 DBG3("UST app recv stream from consumer. Consumer is dead.");
 355                         }
 356                         goto error;
 357                 }
 358
 359                 /* Order is important this is why a list is used. */
 360                 cds_list_add_tail(&stream->list, &ua_chan->streams.head);
 361                 ua_chan->streams.count++;
 362
 363                 DBG2("UST app stream %d received successfully", ua_chan->streams.count);
 364         }
 365
 366         /* This MUST match or else we have a synchronization problem. */
 367         assert(ua_chan->expected_stream_count == ua_chan->streams.count);
 368
 369         /* Wait for confirmation that we can proceed with the streams. */
 370         ret = consumer_recv_status_reply(socket);
 371         if (ret < 0) {
 372                 goto error;
 373         }
 374
 375 error:
 376         health_code_update();
 377         pthread_mutex_unlock(socket->lock);
 378         return ret;
 379 }
 380
 381 /*
 382  * Send a destroy channel command to consumer using the given channel key.
 383  *
 384  * Note that this command MUST be used prior to a successful
 385  * LTTNG_CONSUMER_GET_CHANNEL because once this command is done successfully,
 386  * the streams are dispatched to the consumer threads and MUST be teardown
 387  * through the hang up process.
 388  *
 389  * Return 0 on success else a negative value.
 390  */
 391 int ust_consumer_destroy_channel(struct consumer_socket *socket,
 392                 struct ust_app_channel *ua_chan)
 393 {
 394         int ret;
 395         struct lttcomm_consumer_msg msg;
 396
 397         assert(ua_chan);
 398         assert(socket);
 399
 400         memset(&msg, 0, sizeof(msg));
 401         msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
 402         msg.u.destroy_channel.key = ua_chan->key;
 403
 404         pthread_mutex_lock(socket->lock);
 405         health_code_update();
 406
 407         ret = consumer_send_msg(socket, &msg);
 408         if (ret < 0) {
 409                 goto error;
 410         }
 411
 412 error:
 413         health_code_update();
 414         pthread_mutex_unlock(socket->lock);
 415         return ret;
 416 }
 417
 418 /*
 419  * Send a given stream to UST tracer.
 420  *
 421  * On success return 0 else a negative value.
 422  */
 423 int ust_consumer_send_stream_to_ust(struct ust_app *app,
 424                 struct ust_app_channel *channel, struct ust_app_stream *stream)
 425 {
 426         int ret;
 427
 428         assert(app);
 429         assert(stream);
 430         assert(channel);
 431
 432         DBG2("UST consumer send stream to app %d", app->sock);
 433
 434         /* Relay stream to application. */
 435         pthread_mutex_lock(&app->sock_lock);
 436         ret = ustctl_send_stream_to_ust(app->sock, channel->obj, stream->obj);
 437         pthread_mutex_unlock(&app->sock_lock);
 438         if (ret < 0) {
 439                 if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
 440                         ERR("ustctl send stream handle %d to app pid: %d with ret %d",
 441                                         stream->obj->handle, app->pid, ret);
 442                 } else {
 443                         DBG3("UST app send stream to ust failed. Application is dead.");
 444                 }
 445                 goto error;
 446         }
 447         channel->handle = channel->obj->handle;
 448
 449 error:
 450         return ret;
 451 }
 452
 453 /*
 454  * Send channel previously received from the consumer to the UST tracer.
 455  *
 456  * On success return 0 else a negative value.
 457  */
 458 int ust_consumer_send_channel_to_ust(struct ust_app *app,
 459                 struct ust_app_session *ua_sess, struct ust_app_channel *channel)
 460 {
 461         int ret;
 462
 463         assert(app);
 464         assert(ua_sess);
 465         assert(channel);
 466         assert(channel->obj);
 467
 468         DBG2("UST app send channel to sock %d pid %d (name: %s, key: %" PRIu64 ")",
 469                         app->sock, app->pid, channel->name, channel->tracing_channel_id);
 470
 471         /* Send stream to application. */
 472         pthread_mutex_lock(&app->sock_lock);
 473         ret = ustctl_send_channel_to_ust(app->sock, ua_sess->handle, channel->obj);
 474         pthread_mutex_unlock(&app->sock_lock);
 475         if (ret < 0) {
 476                 if (ret != -EPIPE && ret != -LTTNG_UST_ERR_EXITING) {
 477                         ERR("Error ustctl send channel %s to app pid: %d with ret %d",
 478                                         channel->name, app->pid, ret);
 479                 } else {
 480                         DBG3("UST app send channel to ust failed. Application is dead.");
 481                 }
 482                 goto error;
 483         }
 484
 485 error:
 486         return ret;
 487 }
 488
 489 /*
 490  * Handle the metadata requests from the UST consumer
 491  *
 492  * Return 0 on success else a negative value.
 493  */
 494 int ust_consumer_metadata_request(struct consumer_socket *socket)
 495 {
 496         int ret;
 497         ssize_t ret_push;
 498         struct lttcomm_metadata_request_msg request;
 499         struct buffer_reg_uid *reg_uid;
 500         struct ust_registry_session *ust_reg;
 501         struct lttcomm_consumer_msg msg;
 502
 503         assert(socket);
 504
 505         rcu_read_lock();
 506         health_code_update();
 507
 508         /* Wait for a metadata request */
 509         pthread_mutex_lock(socket->lock);
 510         ret = consumer_socket_recv(socket, &request, sizeof(request));
 511         pthread_mutex_unlock(socket->lock);
 512         if (ret < 0) {
 513                 goto end;
 514         }
 515
 516         DBG("Metadata request received for session %" PRIu64 ", key %" PRIu64,
 517                         request.session_id, request.key);
 518
 519         reg_uid = buffer_reg_uid_find(request.session_id,
 520                         request.bits_per_long, request.uid);
 521         if (reg_uid) {
 522                 ust_reg = reg_uid->registry->reg.ust;
 523         } else {
 524                 struct buffer_reg_pid *reg_pid =
 525                         buffer_reg_pid_find(request.session_id_per_pid);
 526                 if (!reg_pid) {
 527                         DBG("PID registry not found for session id %" PRIu64,
 528                                         request.session_id_per_pid);
 529
 530                         memset(&msg, 0, sizeof(msg));
 531                         msg.cmd_type = LTTNG_ERR_UND;
 532                         pthread_mutex_lock(socket->lock);
 533                         (void) consumer_send_msg(socket, &msg);
 534                         pthread_mutex_unlock(socket->lock);
 535                         /*
 536                          * This is possible since the session might have been destroyed
 537                          * during a consumer metadata request. So here, return gracefully
 538                          * because the destroy session will push the remaining metadata to
 539                          * the consumer.
 540                          */
 541                         ret = 0;
 542                         goto end;
 543                 }
 544                 ust_reg = reg_pid->registry->reg.ust;
 545         }
 546         assert(ust_reg);
 547
 548         pthread_mutex_lock(&ust_reg->lock);
 549         ret_push = ust_app_push_metadata(ust_reg, socket, 1);
 550         pthread_mutex_unlock(&ust_reg->lock);
 551         if (ret_push == -EPIPE) {
 552                 DBG("Application or relay closed while pushing metadata");
 553         } else if (ret_push < 0) {
 554                 ERR("Pushing metadata");
 555                 ret = -1;
 556                 goto end;
 557         } else {
 558                 DBG("UST Consumer metadata pushed successfully");
 559         }
 560         ret = 0;
 561
 562 end:
 563         rcu_read_unlock();
 564         return ret;
 565 }