Fix: return value of register trigger to the notification thread is not acted upon
[lttng-tools.git] / src / bin / lttng-sessiond / kernel-consumer.c
CommitLineData
f1e16794 1/*
ab5be9fa 2 * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
f1e16794 3 *
ab5be9fa 4 * SPDX-License-Identifier: GPL-2.0-only
f1e16794 5 *
f1e16794
DG
6 */
7
6c1c0768 8#define _LGPL_SOURCE
f1e16794
DG
9#include <stdio.h>
10#include <stdlib.h>
f1e16794
DG
11#include <sys/stat.h>
12#include <unistd.h>
e1f3997a 13#include <inttypes.h>
f1e16794
DG
14
15#include <common/common.h>
16#include <common/defaults.h>
f5436bfc 17#include <common/compat/string.h>
f1e16794 18
00e2e675 19#include "consumer.h"
8782cc74 20#include "health-sessiond.h"
f1e16794 21#include "kernel-consumer.h"
e9404c27
JG
22#include "notification-thread-commands.h"
23#include "session.h"
24#include "lttng-sessiond.h"
f1e16794 25
5da88b0f
MD
26static char *create_channel_path(struct consumer_output *consumer,
27 size_t *consumer_path_offset)
00e2e675
DG
28{
29 int ret;
ffe60014 30 char tmp_path[PATH_MAX];
2bba9e53 31 char *pathname = NULL;
00e2e675 32
2bba9e53 33 assert(consumer);
00e2e675 34
ffe60014 35 /* Get the right path name destination */
a5ba6fdd
JG
36 if (consumer->type == CONSUMER_DST_LOCAL ||
37 (consumer->type == CONSUMER_DST_NET &&
38 consumer->relay_major_version == 2 &&
39 consumer->relay_minor_version >= 11)) {
d2956687 40 pathname = strdup(consumer->domain_subdir);
bb3c4e70 41 if (!pathname) {
d2956687
JG
42 PERROR("Failed to copy domain subdirectory string %s",
43 consumer->domain_subdir);
bb3c4e70
MD
44 goto error;
45 }
5da88b0f 46 *consumer_path_offset = strlen(consumer->domain_subdir);
d2956687
JG
47 DBG3("Kernel local consumer trace path relative to current trace chunk: \"%s\"",
48 pathname);
ffe60014 49 } else {
5da88b0f 50 /* Network output, relayd < 2.11. */
2b29c638
JD
51 ret = snprintf(tmp_path, sizeof(tmp_path), "%s%s",
52 consumer->dst.net.base_dir,
b178f53e 53 consumer->domain_subdir);
ffe60014 54 if (ret < 0) {
2bba9e53 55 PERROR("snprintf kernel metadata path");
ffe60014 56 goto error;
dba13f1d
JG
57 } else if (ret >= sizeof(tmp_path)) {
58 ERR("Kernel channel path exceeds the maximal allowed length of of %zu bytes (%i bytes required) with path \"%s%s\"",
59 sizeof(tmp_path), ret,
60 consumer->dst.net.base_dir,
b178f53e 61 consumer->domain_subdir);
dba13f1d 62 goto error;
ffe60014 63 }
f5436bfc 64 pathname = lttng_strndup(tmp_path, sizeof(tmp_path));
bb3c4e70 65 if (!pathname) {
f5436bfc 66 PERROR("lttng_strndup");
bb3c4e70
MD
67 goto error;
68 }
5da88b0f 69 *consumer_path_offset = 0;
ffe60014
DG
70 DBG3("Kernel network consumer subdir path: %s", pathname);
71 }
72
2bba9e53
DG
73 return pathname;
74
75error:
76 free(pathname);
77 return NULL;
78}
79
80/*
81 * Sending a single channel to the consumer with command ADD_CHANNEL.
82 */
ba27cd00 83static
2bba9e53 84int kernel_consumer_add_channel(struct consumer_socket *sock,
e9404c27
JG
85 struct ltt_kernel_channel *channel,
86 struct ltt_kernel_session *ksession,
2bba9e53
DG
87 unsigned int monitor)
88{
89 int ret;
d2956687 90 char *pathname = NULL;
2bba9e53
DG
91 struct lttcomm_consumer_msg lkm;
92 struct consumer_output *consumer;
e9404c27 93 enum lttng_error_code status;
e32d7f27 94 struct ltt_session *session = NULL;
e9404c27 95 struct lttng_channel_extended *channel_attr_extended;
d2956687 96 bool is_local_trace;
5da88b0f 97 size_t consumer_path_offset = 0;
2bba9e53
DG
98
99 /* Safety net */
100 assert(channel);
e9404c27
JG
101 assert(ksession);
102 assert(ksession->consumer);
2bba9e53 103
e9404c27
JG
104 consumer = ksession->consumer;
105 channel_attr_extended = (struct lttng_channel_extended *)
106 channel->channel->attr.extended.ptr;
2bba9e53
DG
107
108 DBG("Kernel consumer adding channel %s to kernel consumer",
109 channel->channel->name);
d2956687 110 is_local_trace = consumer->net_seq_index == -1ULL;
2bba9e53 111
5da88b0f 112 pathname = create_channel_path(consumer, &consumer_path_offset);
bb3c4e70
MD
113 if (!pathname) {
114 ret = -1;
115 goto error;
116 }
2bba9e53 117
d2956687
JG
118 if (is_local_trace && ksession->current_trace_chunk) {
119 enum lttng_trace_chunk_status chunk_status;
120 char *pathname_index;
121
122 ret = asprintf(&pathname_index, "%s/" DEFAULT_INDEX_DIR,
123 pathname);
124 if (ret < 0) {
125 ERR("Failed to format channel index directory");
126 ret = -1;
127 goto error;
128 }
129
130 /*
131 * Create the index subdirectory which will take care
132 * of implicitly creating the channel's path.
133 */
134 chunk_status = lttng_trace_chunk_create_subdirectory(
135 ksession->current_trace_chunk, pathname_index);
136 free(pathname_index);
137 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
138 ret = -1;
139 goto error;
140 }
141 }
142
00e2e675 143 /* Prep channel message structure */
638e7b4e 144 consumer_init_add_channel_comm_msg(&lkm,
e1f3997a 145 channel->key,
e9404c27 146 ksession->id,
5da88b0f 147 &pathname[consumer_path_offset],
e9404c27
JG
148 ksession->uid,
149 ksession->gid,
ffe60014 150 consumer->net_seq_index,
c30aaa51 151 channel->channel->name,
ffe60014
DG
152 channel->stream_count,
153 channel->channel->attr.output,
1624d5b7
JD
154 CONSUMER_CHANNEL_TYPE_DATA,
155 channel->channel->attr.tracefile_size,
2bba9e53 156 channel->channel->attr.tracefile_count,
ecc48a90 157 monitor,
e9404c27 158 channel->channel->attr.live_timer_interval,
a2814ea7 159 ksession->is_live_session,
d2956687
JG
160 channel_attr_extended->monitor_timer_interval,
161 ksession->current_trace_chunk);
00e2e675 162
840cb59c 163 health_code_update();
ca03de58 164
00e2e675
DG
165 ret = consumer_send_channel(sock, &lkm);
166 if (ret < 0) {
167 goto error;
168 }
169
840cb59c 170 health_code_update();
e9404c27
JG
171 rcu_read_lock();
172 session = session_find_by_id(ksession->id);
173 assert(session);
e098433c
JG
174 assert(pthread_mutex_trylock(&session->lock));
175 assert(session_trylock_list());
ca03de58 176
e9404c27
JG
177 status = notification_thread_command_add_channel(
178 notification_thread_handle, session->name,
179 ksession->uid, ksession->gid,
e1f3997a 180 channel->channel->name, channel->key,
e9404c27
JG
181 LTTNG_DOMAIN_KERNEL,
182 channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
183 rcu_read_unlock();
184 if (status != LTTNG_OK) {
185 ret = -1;
186 goto error;
187 }
753873bf
JR
188
189 channel->published_to_notification_thread = true;
190
00e2e675 191error:
e32d7f27
JG
192 if (session) {
193 session_put(session);
194 }
53efb85a 195 free(pathname);
00e2e675
DG
196 return ret;
197}
198
199/*
200 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
9a318688
JG
201 *
202 * The consumer socket lock must be held by the caller.
00e2e675 203 */
f50f23d9 204int kernel_consumer_add_metadata(struct consumer_socket *sock,
e098433c 205 struct ltt_kernel_session *ksession, unsigned int monitor)
00e2e675
DG
206{
207 int ret;
00e2e675 208 struct lttcomm_consumer_msg lkm;
a7d9a3e7 209 struct consumer_output *consumer;
e098433c
JG
210
211 rcu_read_lock();
00e2e675
DG
212
213 /* Safety net */
e098433c
JG
214 assert(ksession);
215 assert(ksession->consumer);
f50f23d9 216 assert(sock);
00e2e675 217
e098433c
JG
218 DBG("Sending metadata %d to kernel consumer",
219 ksession->metadata_stream_fd);
00e2e675
DG
220
221 /* Get consumer output pointer */
e098433c 222 consumer = ksession->consumer;
00e2e675 223
00e2e675 224 /* Prep channel message structure */
d42266a4
JG
225 consumer_init_add_channel_comm_msg(&lkm,
226 ksession->metadata->key,
227 ksession->id,
228 "",
229 ksession->uid,
230 ksession->gid,
231 consumer->net_seq_index,
232 ksession->metadata->conf->name,
233 1,
234 ksession->metadata->conf->attr.output,
235 CONSUMER_CHANNEL_TYPE_METADATA,
236 ksession->metadata->conf->attr.tracefile_size,
237 ksession->metadata->conf->attr.tracefile_count,
238 monitor,
239 ksession->metadata->conf->attr.live_timer_interval,
240 ksession->is_live_session,
241 0,
a2814ea7 242 ksession->current_trace_chunk);
00e2e675 243
840cb59c 244 health_code_update();
ca03de58 245
00e2e675
DG
246 ret = consumer_send_channel(sock, &lkm);
247 if (ret < 0) {
248 goto error;
249 }
250
840cb59c 251 health_code_update();
ca03de58 252
00e2e675 253 /* Prep stream message structure */
e098433c
JG
254 consumer_init_add_stream_comm_msg(&lkm,
255 ksession->metadata->key,
256 ksession->metadata_stream_fd,
d2956687 257 0 /* CPU: 0 for metadata. */);
00e2e675 258
840cb59c 259 health_code_update();
ca03de58 260
00e2e675 261 /* Send stream and file descriptor */
a7d9a3e7 262 ret = consumer_send_stream(sock, consumer, &lkm,
e098433c 263 &ksession->metadata_stream_fd, 1);
00e2e675
DG
264 if (ret < 0) {
265 goto error;
266 }
267
840cb59c 268 health_code_update();
ca03de58 269
00e2e675 270error:
e098433c 271 rcu_read_unlock();
00e2e675
DG
272 return ret;
273}
274
275/*
276 * Sending a single stream to the consumer with command ADD_STREAM.
277 */
5b0e3ccb 278static
f50f23d9 279int kernel_consumer_add_stream(struct consumer_socket *sock,
e098433c
JG
280 struct ltt_kernel_channel *channel,
281 struct ltt_kernel_stream *stream,
d2956687 282 struct ltt_kernel_session *session, unsigned int monitor)
00e2e675
DG
283{
284 int ret;
00e2e675 285 struct lttcomm_consumer_msg lkm;
a7d9a3e7 286 struct consumer_output *consumer;
00e2e675
DG
287
288 assert(channel);
289 assert(stream);
290 assert(session);
291 assert(session->consumer);
f50f23d9 292 assert(sock);
00e2e675
DG
293
294 DBG("Sending stream %d of channel %s to kernel consumer",
295 stream->fd, channel->channel->name);
296
297 /* Get consumer output pointer */
a7d9a3e7 298 consumer = session->consumer;
00e2e675 299
00e2e675 300 /* Prep stream consumer message */
e098433c 301 consumer_init_add_stream_comm_msg(&lkm,
e1f3997a 302 channel->key,
00e2e675 303 stream->fd,
d2956687 304 stream->cpu);
00e2e675 305
840cb59c 306 health_code_update();
ca03de58 307
00e2e675 308 /* Send stream and file descriptor */
a7d9a3e7 309 ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
00e2e675
DG
310 if (ret < 0) {
311 goto error;
312 }
313
840cb59c 314 health_code_update();
ca03de58 315
00e2e675
DG
316error:
317 return ret;
318}
319
a4baae1b
JD
320/*
321 * Sending the notification that all streams were sent with STREAMS_SENT.
322 */
323int kernel_consumer_streams_sent(struct consumer_socket *sock,
324 struct ltt_kernel_session *session, uint64_t channel_key)
325{
326 int ret;
327 struct lttcomm_consumer_msg lkm;
328 struct consumer_output *consumer;
329
330 assert(sock);
331 assert(session);
332
333 DBG("Sending streams_sent");
334 /* Get consumer output pointer */
335 consumer = session->consumer;
336
337 /* Prep stream consumer message */
338 consumer_init_streams_sent_comm_msg(&lkm,
339 LTTNG_CONSUMER_STREAMS_SENT,
340 channel_key, consumer->net_seq_index);
341
342 health_code_update();
343
344 /* Send stream and file descriptor */
345 ret = consumer_send_msg(sock, &lkm);
346 if (ret < 0) {
347 goto error;
348 }
349
350error:
351 return ret;
352}
353
f1e16794
DG
354/*
355 * Send all stream fds of kernel channel to the consumer.
9a318688
JG
356 *
357 * The consumer socket lock must be held by the caller.
f1e16794 358 */
1fc1b7c8 359int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
e098433c 360 struct ltt_kernel_channel *channel, struct ltt_kernel_session *ksession,
2bba9e53 361 unsigned int monitor)
f1e16794 362{
e99f9447 363 int ret = LTTNG_OK;
f1e16794 364 struct ltt_kernel_stream *stream;
00e2e675
DG
365
366 /* Safety net */
367 assert(channel);
e098433c
JG
368 assert(ksession);
369 assert(ksession->consumer);
f50f23d9 370 assert(sock);
00e2e675 371
e098433c
JG
372 rcu_read_lock();
373
00e2e675 374 /* Bail out if consumer is disabled */
e098433c 375 if (!ksession->consumer->enabled) {
f73fabfd 376 ret = LTTNG_OK;
00e2e675
DG
377 goto error;
378 }
f1e16794
DG
379
380 DBG("Sending streams of channel %s to kernel consumer",
381 channel->channel->name);
382
e99f9447 383 if (!channel->sent_to_consumer) {
e098433c 384 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
e99f9447
MD
385 if (ret < 0) {
386 goto error;
387 }
388 channel->sent_to_consumer = true;
f1e16794
DG
389 }
390
391 /* Send streams */
392 cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
6986ab9b 393 if (!stream->fd || stream->sent_to_consumer) {
f1e16794
DG
394 continue;
395 }
00e2e675
DG
396
397 /* Add stream on the kernel consumer side. */
e098433c 398 ret = kernel_consumer_add_stream(sock, channel, stream,
d2956687 399 ksession, monitor);
f1e16794 400 if (ret < 0) {
f1e16794
DG
401 goto error;
402 }
6986ab9b 403 stream->sent_to_consumer = true;
f1e16794
DG
404 }
405
f1e16794 406error:
e098433c 407 rcu_read_unlock();
f1e16794
DG
408 return ret;
409}
410
411/*
412 * Send all stream fds of the kernel session to the consumer.
9a318688
JG
413 *
414 * The consumer socket lock must be held by the caller.
f1e16794 415 */
f50f23d9
DG
416int kernel_consumer_send_session(struct consumer_socket *sock,
417 struct ltt_kernel_session *session)
f1e16794 418{
2bba9e53 419 int ret, monitor = 0;
f1e16794 420 struct ltt_kernel_channel *chan;
f1e16794 421
00e2e675
DG
422 /* Safety net */
423 assert(session);
424 assert(session->consumer);
f50f23d9 425 assert(sock);
f1e16794 426
00e2e675
DG
427 /* Bail out if consumer is disabled */
428 if (!session->consumer->enabled) {
f73fabfd 429 ret = LTTNG_OK;
00e2e675 430 goto error;
f1e16794
DG
431 }
432
2bba9e53
DG
433 /* Don't monitor the streams on the consumer if in flight recorder. */
434 if (session->output_traces) {
435 monitor = 1;
436 }
437
00e2e675
DG
438 DBG("Sending session stream to kernel consumer");
439
609af759 440 if (session->metadata_stream_fd >= 0 && session->metadata) {
2bba9e53 441 ret = kernel_consumer_add_metadata(sock, session, monitor);
f1e16794 442 if (ret < 0) {
f1e16794
DG
443 goto error;
444 }
f1e16794
DG
445 }
446
00e2e675 447 /* Send channel and streams of it */
f1e16794 448 cds_list_for_each_entry(chan, &session->channel_list.head, list) {
1fc1b7c8 449 ret = kernel_consumer_send_channel_streams(sock, chan, session,
2bba9e53 450 monitor);
f1e16794
DG
451 if (ret < 0) {
452 goto error;
453 }
601262d6
JD
454 if (monitor) {
455 /*
456 * Inform the relay that all the streams for the
457 * channel were sent.
458 */
e1f3997a 459 ret = kernel_consumer_streams_sent(sock, session, chan->key);
601262d6
JD
460 if (ret < 0) {
461 goto error;
462 }
463 }
f1e16794
DG
464 }
465
00e2e675 466 DBG("Kernel consumer FDs of metadata and channel streams sent");
f1e16794 467
4ce9ff51 468 session->consumer_fds_sent = 1;
f1e16794
DG
469 return 0;
470
471error:
472 return ret;
473}
07b86b52
JD
474
475int kernel_consumer_destroy_channel(struct consumer_socket *socket,
476 struct ltt_kernel_channel *channel)
477{
478 int ret;
479 struct lttcomm_consumer_msg msg;
480
481 assert(channel);
482 assert(socket);
07b86b52 483
e1f3997a 484 DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key);
07b86b52 485
53efb85a 486 memset(&msg, 0, sizeof(msg));
07b86b52 487 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
e1f3997a 488 msg.u.destroy_channel.key = channel->key;
07b86b52
JD
489
490 pthread_mutex_lock(socket->lock);
491 health_code_update();
492
493 ret = consumer_send_msg(socket, &msg);
494 if (ret < 0) {
495 goto error;
496 }
497
498error:
499 health_code_update();
500 pthread_mutex_unlock(socket->lock);
501 return ret;
502}
503
504int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
505 struct ltt_kernel_metadata *metadata)
506{
507 int ret;
508 struct lttcomm_consumer_msg msg;
509
510 assert(metadata);
511 assert(socket);
07b86b52 512
d40f0359 513 DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key);
07b86b52 514
53efb85a 515 memset(&msg, 0, sizeof(msg));
07b86b52 516 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
d40f0359 517 msg.u.destroy_channel.key = metadata->key;
07b86b52
JD
518
519 pthread_mutex_lock(socket->lock);
520 health_code_update();
521
522 ret = consumer_send_msg(socket, &msg);
523 if (ret < 0) {
524 goto error;
525 }
526
527error:
528 health_code_update();
529 pthread_mutex_unlock(socket->lock);
530 return ret;
531}
This page took 0.08338 seconds and 4 git commands to generate.