Fix: sessiond: abort called on undefined client command
[lttng-tools.git] / src / bin / lttng-sessiond / kernel-consumer.cpp
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 14
c9e313bc
SM
15#include <common/common.hpp>
16#include <common/defaults.hpp>
17#include <common/compat/string.hpp>
f1e16794 18
c9e313bc
SM
19#include "consumer.hpp"
20#include "health-sessiond.hpp"
21#include "kernel-consumer.hpp"
22#include "notification-thread-commands.hpp"
23#include "session.hpp"
24#include "lttng-sessiond.hpp"
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
a0377dfe 33 LTTNG_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 */
a0377dfe
FD
100 LTTNG_ASSERT(channel);
101 LTTNG_ASSERT(ksession);
102 LTTNG_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],
ffe60014 148 consumer->net_seq_index,
c30aaa51 149 channel->channel->name,
ffe60014
DG
150 channel->stream_count,
151 channel->channel->attr.output,
1624d5b7
JD
152 CONSUMER_CHANNEL_TYPE_DATA,
153 channel->channel->attr.tracefile_size,
2bba9e53 154 channel->channel->attr.tracefile_count,
ecc48a90 155 monitor,
e9404c27 156 channel->channel->attr.live_timer_interval,
a2814ea7 157 ksession->is_live_session,
d2956687
JG
158 channel_attr_extended->monitor_timer_interval,
159 ksession->current_trace_chunk);
00e2e675 160
840cb59c 161 health_code_update();
ca03de58 162
00e2e675
DG
163 ret = consumer_send_channel(sock, &lkm);
164 if (ret < 0) {
165 goto error;
166 }
167
840cb59c 168 health_code_update();
e9404c27
JG
169 rcu_read_lock();
170 session = session_find_by_id(ksession->id);
a0377dfe 171 LTTNG_ASSERT(session);
3130a40c
JG
172 ASSERT_LOCKED(session->lock);
173 ASSERT_SESSION_LIST_LOCKED();
ca03de58 174
139a8d25
JG
175 status = notification_thread_command_add_channel(the_notification_thread_handle,
176 session->id, channel->channel->name, channel->key, LTTNG_DOMAIN_KERNEL,
177 channel->channel->attr.subbuf_size * channel->channel->attr.num_subbuf);
e9404c27
JG
178 rcu_read_unlock();
179 if (status != LTTNG_OK) {
180 ret = -1;
181 goto error;
182 }
753873bf
JR
183
184 channel->published_to_notification_thread = true;
185
00e2e675 186error:
e32d7f27
JG
187 if (session) {
188 session_put(session);
189 }
53efb85a 190 free(pathname);
00e2e675
DG
191 return ret;
192}
193
194/*
195 * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
9a318688
JG
196 *
197 * The consumer socket lock must be held by the caller.
00e2e675 198 */
f50f23d9 199int kernel_consumer_add_metadata(struct consumer_socket *sock,
e098433c 200 struct ltt_kernel_session *ksession, unsigned int monitor)
00e2e675
DG
201{
202 int ret;
00e2e675 203 struct lttcomm_consumer_msg lkm;
a7d9a3e7 204 struct consumer_output *consumer;
e098433c
JG
205
206 rcu_read_lock();
00e2e675
DG
207
208 /* Safety net */
a0377dfe
FD
209 LTTNG_ASSERT(ksession);
210 LTTNG_ASSERT(ksession->consumer);
211 LTTNG_ASSERT(sock);
00e2e675 212
e098433c
JG
213 DBG("Sending metadata %d to kernel consumer",
214 ksession->metadata_stream_fd);
00e2e675
DG
215
216 /* Get consumer output pointer */
e098433c 217 consumer = ksession->consumer;
00e2e675 218
00e2e675 219 /* Prep channel message structure */
d42266a4
JG
220 consumer_init_add_channel_comm_msg(&lkm,
221 ksession->metadata->key,
222 ksession->id,
223 "",
d42266a4
JG
224 consumer->net_seq_index,
225 ksession->metadata->conf->name,
226 1,
227 ksession->metadata->conf->attr.output,
228 CONSUMER_CHANNEL_TYPE_METADATA,
229 ksession->metadata->conf->attr.tracefile_size,
230 ksession->metadata->conf->attr.tracefile_count,
231 monitor,
232 ksession->metadata->conf->attr.live_timer_interval,
233 ksession->is_live_session,
234 0,
a2814ea7 235 ksession->current_trace_chunk);
00e2e675 236
840cb59c 237 health_code_update();
ca03de58 238
00e2e675
DG
239 ret = consumer_send_channel(sock, &lkm);
240 if (ret < 0) {
241 goto error;
242 }
243
840cb59c 244 health_code_update();
ca03de58 245
00e2e675 246 /* Prep stream message structure */
e098433c
JG
247 consumer_init_add_stream_comm_msg(&lkm,
248 ksession->metadata->key,
249 ksession->metadata_stream_fd,
d2956687 250 0 /* CPU: 0 for metadata. */);
00e2e675 251
840cb59c 252 health_code_update();
ca03de58 253
00e2e675 254 /* Send stream and file descriptor */
a7d9a3e7 255 ret = consumer_send_stream(sock, consumer, &lkm,
e098433c 256 &ksession->metadata_stream_fd, 1);
00e2e675
DG
257 if (ret < 0) {
258 goto error;
259 }
260
840cb59c 261 health_code_update();
ca03de58 262
00e2e675 263error:
e098433c 264 rcu_read_unlock();
00e2e675
DG
265 return ret;
266}
267
268/*
269 * Sending a single stream to the consumer with command ADD_STREAM.
270 */
5b0e3ccb 271static
f50f23d9 272int kernel_consumer_add_stream(struct consumer_socket *sock,
e098433c
JG
273 struct ltt_kernel_channel *channel,
274 struct ltt_kernel_stream *stream,
f46376a1 275 struct ltt_kernel_session *session)
00e2e675
DG
276{
277 int ret;
00e2e675 278 struct lttcomm_consumer_msg lkm;
a7d9a3e7 279 struct consumer_output *consumer;
00e2e675 280
a0377dfe
FD
281 LTTNG_ASSERT(channel);
282 LTTNG_ASSERT(stream);
283 LTTNG_ASSERT(session);
284 LTTNG_ASSERT(session->consumer);
285 LTTNG_ASSERT(sock);
00e2e675
DG
286
287 DBG("Sending stream %d of channel %s to kernel consumer",
288 stream->fd, channel->channel->name);
289
290 /* Get consumer output pointer */
a7d9a3e7 291 consumer = session->consumer;
00e2e675 292
00e2e675 293 /* Prep stream consumer message */
e098433c 294 consumer_init_add_stream_comm_msg(&lkm,
e1f3997a 295 channel->key,
00e2e675 296 stream->fd,
d2956687 297 stream->cpu);
00e2e675 298
840cb59c 299 health_code_update();
ca03de58 300
00e2e675 301 /* Send stream and file descriptor */
a7d9a3e7 302 ret = consumer_send_stream(sock, consumer, &lkm, &stream->fd, 1);
00e2e675
DG
303 if (ret < 0) {
304 goto error;
305 }
306
840cb59c 307 health_code_update();
ca03de58 308
00e2e675
DG
309error:
310 return ret;
311}
312
a4baae1b
JD
313/*
314 * Sending the notification that all streams were sent with STREAMS_SENT.
315 */
316int kernel_consumer_streams_sent(struct consumer_socket *sock,
317 struct ltt_kernel_session *session, uint64_t channel_key)
318{
319 int ret;
320 struct lttcomm_consumer_msg lkm;
321 struct consumer_output *consumer;
322
a0377dfe
FD
323 LTTNG_ASSERT(sock);
324 LTTNG_ASSERT(session);
a4baae1b
JD
325
326 DBG("Sending streams_sent");
327 /* Get consumer output pointer */
328 consumer = session->consumer;
329
330 /* Prep stream consumer message */
331 consumer_init_streams_sent_comm_msg(&lkm,
332 LTTNG_CONSUMER_STREAMS_SENT,
333 channel_key, consumer->net_seq_index);
334
335 health_code_update();
336
337 /* Send stream and file descriptor */
338 ret = consumer_send_msg(sock, &lkm);
339 if (ret < 0) {
340 goto error;
341 }
342
343error:
344 return ret;
345}
346
f1e16794
DG
347/*
348 * Send all stream fds of kernel channel to the consumer.
9a318688
JG
349 *
350 * The consumer socket lock must be held by the caller.
f1e16794 351 */
1fc1b7c8 352int kernel_consumer_send_channel_streams(struct consumer_socket *sock,
e098433c 353 struct ltt_kernel_channel *channel, struct ltt_kernel_session *ksession,
2bba9e53 354 unsigned int monitor)
f1e16794 355{
e99f9447 356 int ret = LTTNG_OK;
f1e16794 357 struct ltt_kernel_stream *stream;
00e2e675
DG
358
359 /* Safety net */
a0377dfe
FD
360 LTTNG_ASSERT(channel);
361 LTTNG_ASSERT(ksession);
362 LTTNG_ASSERT(ksession->consumer);
363 LTTNG_ASSERT(sock);
00e2e675 364
e098433c
JG
365 rcu_read_lock();
366
00e2e675 367 /* Bail out if consumer is disabled */
e098433c 368 if (!ksession->consumer->enabled) {
f73fabfd 369 ret = LTTNG_OK;
00e2e675
DG
370 goto error;
371 }
f1e16794
DG
372
373 DBG("Sending streams of channel %s to kernel consumer",
374 channel->channel->name);
375
e99f9447 376 if (!channel->sent_to_consumer) {
e098433c 377 ret = kernel_consumer_add_channel(sock, channel, ksession, monitor);
e99f9447
MD
378 if (ret < 0) {
379 goto error;
380 }
381 channel->sent_to_consumer = true;
f1e16794
DG
382 }
383
384 /* Send streams */
385 cds_list_for_each_entry(stream, &channel->stream_list.head, list) {
6986ab9b 386 if (!stream->fd || stream->sent_to_consumer) {
f1e16794
DG
387 continue;
388 }
00e2e675
DG
389
390 /* Add stream on the kernel consumer side. */
e098433c 391 ret = kernel_consumer_add_stream(sock, channel, stream,
f46376a1 392 ksession);
f1e16794 393 if (ret < 0) {
f1e16794
DG
394 goto error;
395 }
6986ab9b 396 stream->sent_to_consumer = true;
f1e16794
DG
397 }
398
f1e16794 399error:
e098433c 400 rcu_read_unlock();
f1e16794
DG
401 return ret;
402}
403
404/*
405 * Send all stream fds of the kernel session to the consumer.
9a318688
JG
406 *
407 * The consumer socket lock must be held by the caller.
f1e16794 408 */
f50f23d9
DG
409int kernel_consumer_send_session(struct consumer_socket *sock,
410 struct ltt_kernel_session *session)
f1e16794 411{
2bba9e53 412 int ret, monitor = 0;
f1e16794 413 struct ltt_kernel_channel *chan;
f1e16794 414
00e2e675 415 /* Safety net */
a0377dfe
FD
416 LTTNG_ASSERT(session);
417 LTTNG_ASSERT(session->consumer);
418 LTTNG_ASSERT(sock);
f1e16794 419
00e2e675
DG
420 /* Bail out if consumer is disabled */
421 if (!session->consumer->enabled) {
f73fabfd 422 ret = LTTNG_OK;
00e2e675 423 goto error;
f1e16794
DG
424 }
425
2bba9e53
DG
426 /* Don't monitor the streams on the consumer if in flight recorder. */
427 if (session->output_traces) {
428 monitor = 1;
429 }
430
00e2e675
DG
431 DBG("Sending session stream to kernel consumer");
432
609af759 433 if (session->metadata_stream_fd >= 0 && session->metadata) {
2bba9e53 434 ret = kernel_consumer_add_metadata(sock, session, monitor);
f1e16794 435 if (ret < 0) {
f1e16794
DG
436 goto error;
437 }
f1e16794
DG
438 }
439
00e2e675 440 /* Send channel and streams of it */
f1e16794 441 cds_list_for_each_entry(chan, &session->channel_list.head, list) {
1fc1b7c8 442 ret = kernel_consumer_send_channel_streams(sock, chan, session,
2bba9e53 443 monitor);
f1e16794
DG
444 if (ret < 0) {
445 goto error;
446 }
601262d6
JD
447 if (monitor) {
448 /*
449 * Inform the relay that all the streams for the
450 * channel were sent.
451 */
e1f3997a 452 ret = kernel_consumer_streams_sent(sock, session, chan->key);
601262d6
JD
453 if (ret < 0) {
454 goto error;
455 }
456 }
f1e16794
DG
457 }
458
00e2e675 459 DBG("Kernel consumer FDs of metadata and channel streams sent");
f1e16794 460
4ce9ff51 461 session->consumer_fds_sent = 1;
f1e16794
DG
462 return 0;
463
464error:
465 return ret;
466}
07b86b52
JD
467
468int kernel_consumer_destroy_channel(struct consumer_socket *socket,
469 struct ltt_kernel_channel *channel)
470{
471 int ret;
472 struct lttcomm_consumer_msg msg;
473
a0377dfe
FD
474 LTTNG_ASSERT(channel);
475 LTTNG_ASSERT(socket);
07b86b52 476
e1f3997a 477 DBG("Sending kernel consumer destroy channel key %" PRIu64, channel->key);
07b86b52 478
53efb85a 479 memset(&msg, 0, sizeof(msg));
07b86b52 480 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
e1f3997a 481 msg.u.destroy_channel.key = channel->key;
07b86b52
JD
482
483 pthread_mutex_lock(socket->lock);
484 health_code_update();
485
486 ret = consumer_send_msg(socket, &msg);
487 if (ret < 0) {
488 goto error;
489 }
490
491error:
492 health_code_update();
493 pthread_mutex_unlock(socket->lock);
494 return ret;
495}
496
497int kernel_consumer_destroy_metadata(struct consumer_socket *socket,
498 struct ltt_kernel_metadata *metadata)
499{
500 int ret;
501 struct lttcomm_consumer_msg msg;
502
a0377dfe
FD
503 LTTNG_ASSERT(metadata);
504 LTTNG_ASSERT(socket);
07b86b52 505
d40f0359 506 DBG("Sending kernel consumer destroy channel key %" PRIu64, metadata->key);
07b86b52 507
53efb85a 508 memset(&msg, 0, sizeof(msg));
07b86b52 509 msg.cmd_type = LTTNG_CONSUMER_DESTROY_CHANNEL;
d40f0359 510 msg.u.destroy_channel.key = metadata->key;
07b86b52
JD
511
512 pthread_mutex_lock(socket->lock);
513 health_code_update();
514
515 ret = consumer_send_msg(socket, &msg);
516 if (ret < 0) {
517 goto error;
518 }
519
520error:
521 health_code_update();
522 pthread_mutex_unlock(socket->lock);
523 return ret;
524}
This page took 0.097407 seconds and 4 git commands to generate.