SoW-2020-0002: Trace Hit Counters: trigger error reporting integration
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
CommitLineData
00e2e675 1/*
ab5be9fa
MJ
2 * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
3 * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
00e2e675 4 *
ab5be9fa 5 * SPDX-License-Identifier: GPL-2.0-only
00e2e675 6 *
00e2e675
DG
7 */
8
6c1c0768 9#define _LGPL_SOURCE
00e2e675
DG
10#include <assert.h>
11#include <stdio.h>
12#include <stdlib.h>
13#include <string.h>
14#include <sys/stat.h>
15#include <sys/types.h>
16#include <unistd.h>
d88aee68 17#include <inttypes.h>
00e2e675
DG
18
19#include <common/common.h>
20#include <common/defaults.h>
21#include <common/uri.h>
d3e2ba59 22#include <common/relayd/relayd.h>
d2956687 23#include <common/string-utils/format.h>
00e2e675
DG
24
25#include "consumer.h"
8782cc74 26#include "health-sessiond.h"
7972aab2 27#include "ust-app.h"
0b2dc8df 28#include "utils.h"
658f12fa 29#include "lttng-sessiond.h"
00e2e675 30
3b967712
MD
31/*
32 * Return allocated full pathname of the session using the consumer trace path
33 * and subdir if available.
34 *
35 * The caller can safely free(3) the returned value. On error, NULL is
36 * returned.
37 */
38char *setup_channel_trace_path(struct consumer_output *consumer,
5da88b0f 39 const char *session_path, size_t *consumer_path_offset)
3b967712
MD
40{
41 int ret;
42 char *pathname;
43
44 assert(consumer);
45 assert(session_path);
46
47 health_code_update();
48
49 /*
50 * Allocate the string ourself to make sure we never exceed
51 * LTTNG_PATH_MAX.
52 */
53 pathname = zmalloc(LTTNG_PATH_MAX);
54 if (!pathname) {
55 goto error;
56 }
57
58 /* Get correct path name destination */
59 if (consumer->type == CONSUMER_DST_NET &&
60 consumer->relay_major_version == 2 &&
61 consumer->relay_minor_version < 11) {
5da88b0f 62 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s/%s/%s",
3b967712
MD
63 consumer->dst.net.base_dir,
64 consumer->chunk_path, consumer->domain_subdir,
65 session_path);
5da88b0f 66 *consumer_path_offset = 0;
3b967712 67 } else {
5da88b0f 68 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s/%s",
3b967712 69 consumer->domain_subdir, session_path);
5da88b0f 70 *consumer_path_offset = strlen(consumer->domain_subdir) + 1;
3b967712
MD
71 }
72 DBG3("Consumer trace path relative to current trace chunk: \"%s\"",
73 pathname);
74 if (ret < 0) {
75 PERROR("Failed to format channel path");
76 goto error;
77 } else if (ret >= LTTNG_PATH_MAX) {
78 ERR("Truncation occurred while formatting channel path");
3b967712
MD
79 goto error;
80 }
81
82 return pathname;
83error:
84 free(pathname);
85 return NULL;
86}
87
52898cb1
DG
88/*
89 * Send a data payload using a given consumer socket of size len.
90 *
91 * The consumer socket lock MUST be acquired before calling this since this
92 * function can change the fd value.
93 *
94 * Return 0 on success else a negative value on error.
95 */
04ed9e10
JG
96int consumer_socket_send(
97 struct consumer_socket *socket, const void *msg, size_t len)
52898cb1
DG
98{
99 int fd;
100 ssize_t size;
101
102 assert(socket);
9363801e 103 assert(socket->fd_ptr);
52898cb1
DG
104 assert(msg);
105
106 /* Consumer socket is invalid. Stopping. */
9363801e 107 fd = *socket->fd_ptr;
52898cb1
DG
108 if (fd < 0) {
109 goto error;
110 }
111
112 size = lttcomm_send_unix_sock(fd, msg, len);
113 if (size < 0) {
114 /* The above call will print a PERROR on error. */
115 DBG("Error when sending data to consumer on sock %d", fd);
116 /*
92db7cdc
DG
117 * At this point, the socket is not usable anymore thus closing it and
118 * setting the file descriptor to -1 so it is not reused.
52898cb1
DG
119 */
120
121 /* This call will PERROR on error. */
122 (void) lttcomm_close_unix_sock(fd);
9363801e 123 *socket->fd_ptr = -1;
52898cb1
DG
124 goto error;
125 }
126
127 return 0;
128
129error:
130 return -1;
131}
132
133/*
134 * Receive a data payload using a given consumer socket of size len.
135 *
136 * The consumer socket lock MUST be acquired before calling this since this
137 * function can change the fd value.
138 *
139 * Return 0 on success else a negative value on error.
140 */
141int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
142{
143 int fd;
144 ssize_t size;
145
146 assert(socket);
9363801e 147 assert(socket->fd_ptr);
52898cb1
DG
148 assert(msg);
149
150 /* Consumer socket is invalid. Stopping. */
9363801e 151 fd = *socket->fd_ptr;
52898cb1
DG
152 if (fd < 0) {
153 goto error;
154 }
155
156 size = lttcomm_recv_unix_sock(fd, msg, len);
157 if (size <= 0) {
158 /* The above call will print a PERROR on error. */
159 DBG("Error when receiving data from the consumer socket %d", fd);
160 /*
92db7cdc
DG
161 * At this point, the socket is not usable anymore thus closing it and
162 * setting the file descriptor to -1 so it is not reused.
52898cb1
DG
163 */
164
165 /* This call will PERROR on error. */
166 (void) lttcomm_close_unix_sock(fd);
9363801e 167 *socket->fd_ptr = -1;
52898cb1
DG
168 goto error;
169 }
170
171 return 0;
172
173error:
174 return -1;
175}
176
f50f23d9
DG
177/*
178 * Receive a reply command status message from the consumer. Consumer socket
179 * lock MUST be acquired before calling this function.
180 *
181 * Return 0 on success, -1 on recv error or a negative lttng error code which
182 * was possibly returned by the consumer.
183 */
184int consumer_recv_status_reply(struct consumer_socket *sock)
185{
186 int ret;
187 struct lttcomm_consumer_status_msg reply;
188
189 assert(sock);
190
52898cb1
DG
191 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
192 if (ret < 0) {
f50f23d9
DG
193 goto end;
194 }
195
0c759fc9 196 if (reply.ret_code == LTTCOMM_CONSUMERD_SUCCESS) {
f50f23d9
DG
197 /* All good. */
198 ret = 0;
199 } else {
200 ret = -reply.ret_code;
ffe60014 201 DBG("Consumer ret code %d", ret);
f50f23d9
DG
202 }
203
204end:
205 return ret;
206}
207
ffe60014
DG
208/*
209 * Once the ASK_CHANNEL command is sent to the consumer, the channel
210 * information are sent back. This call receives that data and populates key
211 * and stream_count.
212 *
213 * On success return 0 and both key and stream_count are set. On error, a
214 * negative value is sent back and both parameters are untouched.
215 */
216int consumer_recv_status_channel(struct consumer_socket *sock,
d88aee68 217 uint64_t *key, unsigned int *stream_count)
ffe60014
DG
218{
219 int ret;
220 struct lttcomm_consumer_status_channel reply;
221
222 assert(sock);
223 assert(stream_count);
224 assert(key);
225
52898cb1
DG
226 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
227 if (ret < 0) {
ffe60014
DG
228 goto end;
229 }
230
231 /* An error is possible so don't touch the key and stream_count. */
0c759fc9 232 if (reply.ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
ffe60014
DG
233 ret = -1;
234 goto end;
235 }
236
237 *key = reply.key;
238 *stream_count = reply.stream_count;
0c759fc9 239 ret = 0;
ffe60014
DG
240
241end:
242 return ret;
243}
244
2f77fc4b
DG
245/*
246 * Send destroy relayd command to consumer.
247 *
248 * On success return positive value. On error, negative value.
249 */
250int consumer_send_destroy_relayd(struct consumer_socket *sock,
251 struct consumer_output *consumer)
252{
253 int ret;
254 struct lttcomm_consumer_msg msg;
255
256 assert(consumer);
257 assert(sock);
258
9363801e 259 DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
2f77fc4b 260
53efb85a 261 memset(&msg, 0, sizeof(msg));
2f77fc4b
DG
262 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
263 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
264
265 pthread_mutex_lock(sock->lock);
52898cb1 266 ret = consumer_socket_send(sock, &msg, sizeof(msg));
2f77fc4b 267 if (ret < 0) {
52898cb1 268 goto error;
2f77fc4b
DG
269 }
270
f50f23d9
DG
271 /* Don't check the return value. The caller will do it. */
272 ret = consumer_recv_status_reply(sock);
273
2f77fc4b
DG
274 DBG2("Consumer send destroy relayd command done");
275
276error:
52898cb1 277 pthread_mutex_unlock(sock->lock);
2f77fc4b
DG
278 return ret;
279}
280
281/*
282 * For each consumer socket in the consumer output object, send a destroy
283 * relayd command.
284 */
285void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
286{
2f77fc4b
DG
287 struct lttng_ht_iter iter;
288 struct consumer_socket *socket;
289
290 assert(consumer);
291
292 /* Destroy any relayd connection */
6dc3064a 293 if (consumer->type == CONSUMER_DST_NET) {
2f77fc4b
DG
294 rcu_read_lock();
295 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
296 node.node) {
c617c0c6
MD
297 int ret;
298
2f77fc4b
DG
299 /* Send destroy relayd command */
300 ret = consumer_send_destroy_relayd(socket, consumer);
301 if (ret < 0) {
c5c45efa 302 DBG("Unable to send destroy relayd command to consumer");
2f77fc4b
DG
303 /* Continue since we MUST delete everything at this point. */
304 }
305 }
306 rcu_read_unlock();
307 }
308}
309
a4b92340
DG
310/*
311 * From a consumer_data structure, allocate and add a consumer socket to the
312 * consumer output.
313 *
314 * Return 0 on success, else negative value on error
315 */
316int consumer_create_socket(struct consumer_data *data,
317 struct consumer_output *output)
318{
319 int ret = 0;
320 struct consumer_socket *socket;
321
322 assert(data);
323
324 if (output == NULL || data->cmd_sock < 0) {
325 /*
326 * Not an error. Possible there is simply not spawned consumer or it's
327 * disabled for the tracing session asking the socket.
328 */
329 goto error;
330 }
331
332 rcu_read_lock();
333 socket = consumer_find_socket(data->cmd_sock, output);
334 rcu_read_unlock();
335 if (socket == NULL) {
4ce514c4 336 socket = consumer_allocate_socket(&data->cmd_sock);
a4b92340
DG
337 if (socket == NULL) {
338 ret = -1;
339 goto error;
340 }
341
2f77fc4b 342 socket->registered = 0;
a4b92340
DG
343 socket->lock = &data->lock;
344 rcu_read_lock();
345 consumer_add_socket(socket, output);
346 rcu_read_unlock();
347 }
348
6dc3064a
DG
349 socket->type = data->type;
350
a4b92340
DG
351 DBG3("Consumer socket created (fd: %d) and added to output",
352 data->cmd_sock);
353
354error:
355 return ret;
356}
357
7972aab2
DG
358/*
359 * Return the consumer socket from the given consumer output with the right
360 * bitness. On error, returns NULL.
361 *
362 * The caller MUST acquire a rcu read side lock and keep it until the socket
363 * object reference is not needed anymore.
364 */
365struct consumer_socket *consumer_find_socket_by_bitness(int bits,
348a81dc 366 const struct consumer_output *consumer)
7972aab2
DG
367{
368 int consumer_fd;
369 struct consumer_socket *socket = NULL;
370
371 switch (bits) {
372 case 64:
373 consumer_fd = uatomic_read(&ust_consumerd64_fd);
374 break;
375 case 32:
376 consumer_fd = uatomic_read(&ust_consumerd32_fd);
377 break;
378 default:
379 assert(0);
380 goto end;
381 }
382
383 socket = consumer_find_socket(consumer_fd, consumer);
384 if (!socket) {
385 ERR("Consumer socket fd %d not found in consumer obj %p",
386 consumer_fd, consumer);
387 }
388
389end:
390 return socket;
391}
392
173af62f
DG
393/*
394 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
395 * be acquired before calling this function and across use of the
396 * returned consumer_socket.
397 */
398struct consumer_socket *consumer_find_socket(int key,
348a81dc 399 const struct consumer_output *consumer)
173af62f
DG
400{
401 struct lttng_ht_iter iter;
402 struct lttng_ht_node_ulong *node;
403 struct consumer_socket *socket = NULL;
404
405 /* Negative keys are lookup failures */
a4b92340 406 if (key < 0 || consumer == NULL) {
173af62f
DG
407 return NULL;
408 }
409
410 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
411 &iter);
412 node = lttng_ht_iter_get_node_ulong(&iter);
413 if (node != NULL) {
414 socket = caa_container_of(node, struct consumer_socket, node);
415 }
416
417 return socket;
418}
419
420/*
421 * Allocate a new consumer_socket and return the pointer.
422 */
4ce514c4 423struct consumer_socket *consumer_allocate_socket(int *fd)
173af62f
DG
424{
425 struct consumer_socket *socket = NULL;
426
4ce514c4
DG
427 assert(fd);
428
173af62f
DG
429 socket = zmalloc(sizeof(struct consumer_socket));
430 if (socket == NULL) {
431 PERROR("zmalloc consumer socket");
432 goto error;
433 }
434
9363801e 435 socket->fd_ptr = fd;
4ce514c4 436 lttng_ht_node_init_ulong(&socket->node, *fd);
173af62f
DG
437
438error:
439 return socket;
440}
441
442/*
443 * Add consumer socket to consumer output object. Read side lock must be
444 * acquired before calling this function.
445 */
446void consumer_add_socket(struct consumer_socket *sock,
447 struct consumer_output *consumer)
448{
449 assert(sock);
450 assert(consumer);
451
452 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
453}
454
455/*
348a81dc 456 * Delete consumer socket to consumer output object. Read side lock must be
173af62f
DG
457 * acquired before calling this function.
458 */
459void consumer_del_socket(struct consumer_socket *sock,
460 struct consumer_output *consumer)
461{
462 int ret;
463 struct lttng_ht_iter iter;
464
465 assert(sock);
466 assert(consumer);
467
468 iter.iter.node = &sock->node.node;
469 ret = lttng_ht_del(consumer->socks, &iter);
470 assert(!ret);
471}
472
473/*
474 * RCU destroy call function.
475 */
476static void destroy_socket_rcu(struct rcu_head *head)
477{
478 struct lttng_ht_node_ulong *node =
479 caa_container_of(head, struct lttng_ht_node_ulong, head);
480 struct consumer_socket *socket =
481 caa_container_of(node, struct consumer_socket, node);
482
483 free(socket);
484}
485
486/*
487 * Destroy and free socket pointer in a call RCU. Read side lock must be
488 * acquired before calling this function.
489 */
490void consumer_destroy_socket(struct consumer_socket *sock)
491{
492 assert(sock);
493
494 /*
495 * We DO NOT close the file descriptor here since it is global to the
2f77fc4b
DG
496 * session daemon and is closed only if the consumer dies or a custom
497 * consumer was registered,
173af62f 498 */
2f77fc4b 499 if (sock->registered) {
9363801e
DG
500 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
501 lttcomm_close_unix_sock(*sock->fd_ptr);
2f77fc4b 502 }
173af62f
DG
503
504 call_rcu(&sock->node.head, destroy_socket_rcu);
505}
506
00e2e675
DG
507/*
508 * Allocate and assign data to a consumer_output object.
509 *
510 * Return pointer to structure.
511 */
512struct consumer_output *consumer_create_output(enum consumer_dst_type type)
513{
514 struct consumer_output *output = NULL;
515
516 output = zmalloc(sizeof(struct consumer_output));
517 if (output == NULL) {
518 PERROR("zmalloc consumer_output");
519 goto error;
520 }
521
522 /* By default, consumer output is enabled */
523 output->enabled = 1;
524 output->type = type;
d88aee68 525 output->net_seq_index = (uint64_t) -1ULL;
6addfa37 526 urcu_ref_init(&output->ref);
173af62f
DG
527
528 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675
DG
529
530error:
531 return output;
532}
533
af706bb7
DG
534/*
535 * Iterate over the consumer output socket hash table and destroy them. The
536 * socket file descriptor are only closed if the consumer output was
537 * registered meaning it's an external consumer.
538 */
539void consumer_destroy_output_sockets(struct consumer_output *obj)
540{
541 struct lttng_ht_iter iter;
542 struct consumer_socket *socket;
543
544 if (!obj->socks) {
545 return;
546 }
547
548 rcu_read_lock();
549 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
550 consumer_del_socket(socket, obj);
551 consumer_destroy_socket(socket);
552 }
553 rcu_read_unlock();
554}
555
00e2e675
DG
556/*
557 * Delete the consumer_output object from the list and free the ptr.
36b588ed
MD
558 *
559 * Should *NOT* be called with RCU read-side lock held.
00e2e675 560 */
6addfa37 561static void consumer_release_output(struct urcu_ref *ref)
00e2e675 562{
6addfa37
MD
563 struct consumer_output *obj =
564 caa_container_of(ref, struct consumer_output, ref);
00e2e675 565
af706bb7 566 consumer_destroy_output_sockets(obj);
2f77fc4b 567
af706bb7 568 if (obj->socks) {
2f77fc4b 569 /* Finally destroy HT */
0b2dc8df 570 ht_cleanup_push(obj->socks);
00e2e675 571 }
173af62f 572
00e2e675
DG
573 free(obj);
574}
575
6addfa37
MD
576/*
577 * Get the consumer_output object.
578 */
579void consumer_output_get(struct consumer_output *obj)
580{
581 urcu_ref_get(&obj->ref);
582}
583
584/*
585 * Put the consumer_output object.
586 *
587 * Should *NOT* be called with RCU read-side lock held.
588 */
589void consumer_output_put(struct consumer_output *obj)
590{
591 if (!obj) {
592 return;
593 }
594 urcu_ref_put(&obj->ref, consumer_release_output);
595}
596
00e2e675
DG
597/*
598 * Copy consumer output and returned the newly allocated copy.
36b588ed
MD
599 *
600 * Should *NOT* be called with RCU read-side lock held.
00e2e675 601 */
b178f53e 602struct consumer_output *consumer_copy_output(struct consumer_output *src)
00e2e675 603{
6dc3064a 604 int ret;
00e2e675
DG
605 struct consumer_output *output;
606
b178f53e 607 assert(src);
00e2e675 608
b178f53e 609 output = consumer_create_output(src->type);
00e2e675 610 if (output == NULL) {
6addfa37 611 goto end;
00e2e675 612 }
b178f53e
JG
613 output->enabled = src->enabled;
614 output->net_seq_index = src->net_seq_index;
615 memcpy(output->domain_subdir, src->domain_subdir,
616 sizeof(output->domain_subdir));
617 output->snapshot = src->snapshot;
618 output->relay_major_version = src->relay_major_version;
619 output->relay_minor_version = src->relay_minor_version;
eacb7b6f 620 output->relay_allows_clear = src->relay_allows_clear;
b178f53e
JG
621 memcpy(&output->dst, &src->dst, sizeof(output->dst));
622 ret = consumer_copy_sockets(output, src);
6dc3064a 623 if (ret < 0) {
6addfa37 624 goto error_put;
6dc3064a 625 }
6addfa37 626end:
6dc3064a
DG
627 return output;
628
6addfa37
MD
629error_put:
630 consumer_output_put(output);
6dc3064a
DG
631 return NULL;
632}
633
634/*
635 * Copy consumer sockets from src to dst.
636 *
637 * Return 0 on success or else a negative value.
638 */
639int consumer_copy_sockets(struct consumer_output *dst,
640 struct consumer_output *src)
641{
642 int ret = 0;
643 struct lttng_ht_iter iter;
644 struct consumer_socket *socket, *copy_sock;
645
646 assert(dst);
647 assert(src);
648
b82c5c4d 649 rcu_read_lock();
6dc3064a
DG
650 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
651 /* Ignore socket that are already there. */
9363801e 652 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
6dc3064a
DG
653 if (copy_sock) {
654 continue;
655 }
656
173af62f 657 /* Create new socket object. */
9363801e 658 copy_sock = consumer_allocate_socket(socket->fd_ptr);
173af62f 659 if (copy_sock == NULL) {
b82c5c4d 660 rcu_read_unlock();
6dc3064a
DG
661 ret = -ENOMEM;
662 goto error;
173af62f
DG
663 }
664
09a90bcd 665 copy_sock->registered = socket->registered;
6dc3064a
DG
666 /*
667 * This is valid because this lock is shared accross all consumer
668 * object being the global lock of the consumer data structure of the
669 * session daemon.
670 */
173af62f 671 copy_sock->lock = socket->lock;
6dc3064a 672 consumer_add_socket(copy_sock, dst);
173af62f 673 }
b82c5c4d 674 rcu_read_unlock();
173af62f 675
00e2e675 676error:
6dc3064a 677 return ret;
00e2e675
DG
678}
679
680/*
b178f53e 681 * Set network URI to the consumer output.
00e2e675 682 *
ad20f474
DG
683 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
684 * error.
00e2e675 685 */
b178f53e
JG
686int consumer_set_network_uri(const struct ltt_session *session,
687 struct consumer_output *output,
00e2e675
DG
688 struct lttng_uri *uri)
689{
690 int ret;
00e2e675
DG
691 struct lttng_uri *dst_uri = NULL;
692
693 /* Code flow error safety net. */
b178f53e 694 assert(output);
00e2e675
DG
695 assert(uri);
696
697 switch (uri->stype) {
698 case LTTNG_STREAM_CONTROL:
b178f53e
JG
699 dst_uri = &output->dst.net.control;
700 output->dst.net.control_isset = 1;
00e2e675
DG
701 if (uri->port == 0) {
702 /* Assign default port. */
703 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
a74934ba 704 } else {
b178f53e
JG
705 if (output->dst.net.data_isset && uri->port ==
706 output->dst.net.data.port) {
a74934ba
DG
707 ret = -LTTNG_ERR_INVALID;
708 goto error;
709 }
00e2e675 710 }
ad20f474 711 DBG3("Consumer control URI set with port %d", uri->port);
00e2e675
DG
712 break;
713 case LTTNG_STREAM_DATA:
b178f53e
JG
714 dst_uri = &output->dst.net.data;
715 output->dst.net.data_isset = 1;
00e2e675
DG
716 if (uri->port == 0) {
717 /* Assign default port. */
718 uri->port = DEFAULT_NETWORK_DATA_PORT;
a74934ba 719 } else {
b178f53e
JG
720 if (output->dst.net.control_isset && uri->port ==
721 output->dst.net.control.port) {
a74934ba
DG
722 ret = -LTTNG_ERR_INVALID;
723 goto error;
724 }
00e2e675 725 }
ad20f474 726 DBG3("Consumer data URI set with port %d", uri->port);
00e2e675
DG
727 break;
728 default:
729 ERR("Set network uri type unknown %d", uri->stype);
a74934ba 730 ret = -LTTNG_ERR_INVALID;
00e2e675
DG
731 goto error;
732 }
733
734 ret = uri_compare(dst_uri, uri);
735 if (!ret) {
736 /* Same URI, don't touch it and return success. */
737 DBG3("URI network compare are the same");
ad20f474 738 goto equal;
00e2e675
DG
739 }
740
741 /* URIs were not equal, replacing it. */
00e2e675 742 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
b178f53e
JG
743 output->type = CONSUMER_DST_NET;
744 if (dst_uri->stype != LTTNG_STREAM_CONTROL) {
745 /* Only the control uri needs to contain the path. */
746 goto end;
747 }
00e2e675 748
b178f53e
JG
749 /*
750 * If the user has specified a subdir as part of the control
751 * URL, the session's base output directory is:
752 * /RELAYD_OUTPUT_PATH/HOSTNAME/USER_SPECIFIED_DIR
753 *
754 * Hence, the "base_dir" from which all stream files and
755 * session rotation chunks are created takes the form
756 * /HOSTNAME/USER_SPECIFIED_DIR
757 *
758 * If the user has not specified an output directory as part of
759 * the control URL, the base output directory has the form:
760 * /RELAYD_OUTPUT_PATH/HOSTNAME/SESSION_NAME-CREATION_TIME
761 *
762 * Hence, the "base_dir" from which all stream files and
763 * session rotation chunks are created takes the form
764 * /HOSTNAME/SESSION_NAME-CREATION_TIME
765 *
766 * Note that automatically generated session names already
767 * contain the session's creation time. In that case, the
768 * creation time is omitted to prevent it from being duplicated
769 * in the final directory hierarchy.
770 */
771 if (*uri->subdir) {
772 if (strstr(uri->subdir, "../")) {
773 ERR("Network URI subdirs are not allowed to walk up the path hierarchy");
774 ret = -LTTNG_ERR_INVALID;
00e2e675
DG
775 goto error;
776 }
b178f53e
JG
777 ret = snprintf(output->dst.net.base_dir,
778 sizeof(output->dst.net.base_dir),
779 "/%s/%s/", session->hostname, uri->subdir);
780 } else {
781 if (session->has_auto_generated_name) {
782 ret = snprintf(output->dst.net.base_dir,
783 sizeof(output->dst.net.base_dir),
784 "/%s/%s/", session->hostname,
785 session->name);
786 } else {
787 char session_creation_datetime[16];
788 size_t strftime_ret;
789 struct tm *timeinfo;
00e2e675 790
b178f53e
JG
791 timeinfo = localtime(&session->creation_time);
792 if (!timeinfo) {
793 ret = -LTTNG_ERR_FATAL;
794 goto error;
795 }
796 strftime_ret = strftime(session_creation_datetime,
797 sizeof(session_creation_datetime),
798 "%Y%m%d-%H%M%S", timeinfo);
799 if (strftime_ret == 0) {
800 ERR("Failed to format session creation timestamp while setting network URI");
801 ret = -LTTNG_ERR_FATAL;
802 goto error;
803 }
804 ret = snprintf(output->dst.net.base_dir,
805 sizeof(output->dst.net.base_dir),
806 "/%s/%s-%s/", session->hostname,
807 session->name,
808 session_creation_datetime);
bfc6eff0 809 }
00e2e675 810 }
b178f53e
JG
811 if (ret >= sizeof(output->dst.net.base_dir)) {
812 ret = -LTTNG_ERR_INVALID;
813 ERR("Truncation occurred while setting network output base directory");
814 goto error;
815 } else if (ret == -1) {
816 ret = -LTTNG_ERR_INVALID;
817 PERROR("Error occurred while setting network output base directory");
818 goto error;
819 }
820
821 DBG3("Consumer set network uri base_dir path %s",
822 output->dst.net.base_dir);
00e2e675 823
b178f53e 824end:
00e2e675 825 return 0;
ad20f474
DG
826equal:
827 return 1;
00e2e675 828error:
a74934ba 829 return ret;
00e2e675
DG
830}
831
832/*
833 * Send file descriptor to consumer via sock.
9a318688
JG
834 *
835 * The consumer socket lock must be held by the caller.
00e2e675 836 */
ac2f30af
JG
837int consumer_send_fds(struct consumer_socket *sock, const int *fds,
838 size_t nb_fd)
00e2e675
DG
839{
840 int ret;
841
842 assert(fds);
f50f23d9 843 assert(sock);
00e2e675 844 assert(nb_fd > 0);
05dac598 845 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
00e2e675 846
9363801e 847 ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
00e2e675 848 if (ret < 0) {
3448e266 849 /* The above call will print a PERROR on error. */
9363801e 850 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
00e2e675
DG
851 goto error;
852 }
853
f50f23d9 854 ret = consumer_recv_status_reply(sock);
00e2e675
DG
855error:
856 return ret;
857}
858
ffe60014
DG
859/*
860 * Consumer send communication message structure to consumer.
9a318688
JG
861 *
862 * The consumer socket lock must be held by the caller.
ffe60014
DG
863 */
864int consumer_send_msg(struct consumer_socket *sock,
04ed9e10 865 const struct lttcomm_consumer_msg *msg)
ffe60014
DG
866{
867 int ret;
868
869 assert(msg);
870 assert(sock);
211b734b 871 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
ffe60014 872
52898cb1 873 ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
ffe60014 874 if (ret < 0) {
ffe60014
DG
875 goto error;
876 }
877
878 ret = consumer_recv_status_reply(sock);
879
880error:
881 return ret;
882}
883
00e2e675
DG
884/*
885 * Consumer send channel communication message structure to consumer.
9a318688
JG
886 *
887 * The consumer socket lock must be held by the caller.
00e2e675 888 */
f50f23d9
DG
889int consumer_send_channel(struct consumer_socket *sock,
890 struct lttcomm_consumer_msg *msg)
00e2e675
DG
891{
892 int ret;
893
894 assert(msg);
f50f23d9 895 assert(sock);
00e2e675 896
52898cb1 897 ret = consumer_send_msg(sock, msg);
00e2e675 898 if (ret < 0) {
00e2e675
DG
899 goto error;
900 }
901
902error:
903 return ret;
904}
905
ffe60014
DG
906/*
907 * Populate the given consumer msg structure with the ask_channel command
908 * information.
909 */
910void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
911 uint64_t subbuf_size,
912 uint64_t num_subbuf,
913 int overwrite,
914 unsigned int switch_timer_interval,
915 unsigned int read_timer_interval,
ecc48a90 916 unsigned int live_timer_interval,
a2814ea7 917 bool is_in_live_session,
e9404c27 918 unsigned int monitor_timer_interval,
ffe60014
DG
919 int output,
920 int type,
921 uint64_t session_id,
922 const char *pathname,
923 const char *name,
d88aee68
DG
924 uint64_t relayd_id,
925 uint64_t key,
7972aab2 926 unsigned char *uuid,
1624d5b7
JD
927 uint32_t chan_id,
928 uint64_t tracefile_size,
2bba9e53 929 uint64_t tracefile_count,
1950109e 930 uint64_t session_id_per_pid,
567eb353 931 unsigned int monitor,
d7ba1388 932 uint32_t ust_app_uid,
491d1539 933 int64_t blocking_timeout,
3d071855 934 const char *root_shm_path,
e098433c 935 const char *shm_path,
1b0bebcb
JG
936 struct lttng_trace_chunk *trace_chunk,
937 const struct lttng_credentials *buffer_credentials)
ffe60014
DG
938{
939 assert(msg);
940
26c468bb 941 /* Zeroed structure */
ffe60014 942 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
d2956687
JG
943 msg->u.ask_channel.buffer_credentials.uid = UINT32_MAX;
944 msg->u.ask_channel.buffer_credentials.gid = UINT32_MAX;
945
26c468bb 946 if (trace_chunk) {
d2956687
JG
947 uint64_t chunk_id;
948 enum lttng_trace_chunk_status chunk_status;
d2956687
JG
949
950 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
951 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
952 LTTNG_OPTIONAL_SET(&msg->u.ask_channel.chunk_id, chunk_id);
26c468bb 953 }
2463b787
JR
954 msg->u.ask_channel.buffer_credentials.uid =
955 lttng_credentials_get_uid(buffer_credentials);
956 msg->u.ask_channel.buffer_credentials.gid =
957 lttng_credentials_get_gid(buffer_credentials);
ffe60014
DG
958
959 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
960 msg->u.ask_channel.subbuf_size = subbuf_size;
961 msg->u.ask_channel.num_subbuf = num_subbuf ;
962 msg->u.ask_channel.overwrite = overwrite;
963 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
964 msg->u.ask_channel.read_timer_interval = read_timer_interval;
ecc48a90 965 msg->u.ask_channel.live_timer_interval = live_timer_interval;
a2814ea7 966 msg->u.ask_channel.is_live = is_in_live_session;
e9404c27 967 msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
ffe60014
DG
968 msg->u.ask_channel.output = output;
969 msg->u.ask_channel.type = type;
970 msg->u.ask_channel.session_id = session_id;
1950109e 971 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
ffe60014
DG
972 msg->u.ask_channel.relayd_id = relayd_id;
973 msg->u.ask_channel.key = key;
7972aab2 974 msg->u.ask_channel.chan_id = chan_id;
1624d5b7
JD
975 msg->u.ask_channel.tracefile_size = tracefile_size;
976 msg->u.ask_channel.tracefile_count = tracefile_count;
2bba9e53 977 msg->u.ask_channel.monitor = monitor;
567eb353 978 msg->u.ask_channel.ust_app_uid = ust_app_uid;
491d1539 979 msg->u.ask_channel.blocking_timeout = blocking_timeout;
ffe60014
DG
980
981 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
982
10a50311
JD
983 if (pathname) {
984 strncpy(msg->u.ask_channel.pathname, pathname,
985 sizeof(msg->u.ask_channel.pathname));
986 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
987 }
ffe60014
DG
988
989 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
990 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
d7ba1388 991
3d071855
MD
992 if (root_shm_path) {
993 strncpy(msg->u.ask_channel.root_shm_path, root_shm_path,
994 sizeof(msg->u.ask_channel.root_shm_path));
995 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] = '\0';
996 }
d7ba1388
MD
997 if (shm_path) {
998 strncpy(msg->u.ask_channel.shm_path, shm_path,
999 sizeof(msg->u.ask_channel.shm_path));
1000 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
1001 }
ffe60014
DG
1002}
1003
00e2e675
DG
1004/*
1005 * Init channel communication message structure.
1006 */
638e7b4e 1007void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg,
d88aee68 1008 uint64_t channel_key,
ffe60014
DG
1009 uint64_t session_id,
1010 const char *pathname,
1011 uid_t uid,
1012 gid_t gid,
d88aee68 1013 uint64_t relayd_id,
c30aaa51 1014 const char *name,
ffe60014
DG
1015 unsigned int nb_init_streams,
1016 enum lttng_event_output output,
1624d5b7
JD
1017 int type,
1018 uint64_t tracefile_size,
2bba9e53 1019 uint64_t tracefile_count,
ecc48a90 1020 unsigned int monitor,
e9404c27 1021 unsigned int live_timer_interval,
a2814ea7 1022 bool is_in_live_session,
d2956687
JG
1023 unsigned int monitor_timer_interval,
1024 struct lttng_trace_chunk *trace_chunk)
00e2e675
DG
1025{
1026 assert(msg);
1027
00e2e675
DG
1028 /* Zeroed structure */
1029 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1030
26c468bb 1031 if (trace_chunk) {
d2956687
JG
1032 uint64_t chunk_id;
1033 enum lttng_trace_chunk_status chunk_status;
1034
1035 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
1036 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
1037 LTTNG_OPTIONAL_SET(&msg->u.channel.chunk_id, chunk_id);
26c468bb 1038 }
d2956687 1039
00e2e675 1040 /* Send channel */
638e7b4e 1041 msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
00e2e675 1042 msg->u.channel.channel_key = channel_key;
ffe60014 1043 msg->u.channel.session_id = session_id;
ffe60014 1044 msg->u.channel.relayd_id = relayd_id;
c30aaa51 1045 msg->u.channel.nb_init_streams = nb_init_streams;
ffe60014
DG
1046 msg->u.channel.output = output;
1047 msg->u.channel.type = type;
1624d5b7
JD
1048 msg->u.channel.tracefile_size = tracefile_size;
1049 msg->u.channel.tracefile_count = tracefile_count;
2bba9e53 1050 msg->u.channel.monitor = monitor;
ecc48a90 1051 msg->u.channel.live_timer_interval = live_timer_interval;
a2814ea7 1052 msg->u.channel.is_live = is_in_live_session;
e9404c27 1053 msg->u.channel.monitor_timer_interval = monitor_timer_interval;
ffe60014
DG
1054
1055 strncpy(msg->u.channel.pathname, pathname,
1056 sizeof(msg->u.channel.pathname));
1057 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
1058
1059 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
1060 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
00e2e675
DG
1061}
1062
1063/*
1064 * Init stream communication message structure.
1065 */
e098433c 1066void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg,
d88aee68
DG
1067 uint64_t channel_key,
1068 uint64_t stream_key,
d2956687 1069 int32_t cpu)
00e2e675
DG
1070{
1071 assert(msg);
1072
1073 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1074
e098433c 1075 msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM;
00e2e675
DG
1076 msg->u.stream.channel_key = channel_key;
1077 msg->u.stream.stream_key = stream_key;
ffe60014 1078 msg->u.stream.cpu = cpu;
00e2e675
DG
1079}
1080
a4baae1b
JD
1081void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
1082 enum lttng_consumer_command cmd,
1083 uint64_t channel_key, uint64_t net_seq_idx)
1084{
1085 assert(msg);
1086
1087 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1088
1089 msg->cmd_type = cmd;
1090 msg->u.sent_streams.channel_key = channel_key;
1091 msg->u.sent_streams.net_seq_idx = net_seq_idx;
1092}
1093
00e2e675
DG
1094/*
1095 * Send stream communication structure to the consumer.
1096 */
f50f23d9
DG
1097int consumer_send_stream(struct consumer_socket *sock,
1098 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
ac2f30af 1099 const int *fds, size_t nb_fd)
00e2e675
DG
1100{
1101 int ret;
1102
1103 assert(msg);
1104 assert(dst);
f50f23d9 1105 assert(sock);
ffe60014 1106 assert(fds);
00e2e675 1107
52898cb1 1108 ret = consumer_send_msg(sock, msg);
f50f23d9
DG
1109 if (ret < 0) {
1110 goto error;
1111 }
1112
00e2e675
DG
1113 ret = consumer_send_fds(sock, fds, nb_fd);
1114 if (ret < 0) {
1115 goto error;
1116 }
1117
1118error:
1119 return ret;
1120}
37278a1e
DG
1121
1122/*
1123 * Send relayd socket to consumer associated with a session name.
1124 *
43fade62
JG
1125 * The consumer socket lock must be held by the caller.
1126 *
37278a1e
DG
1127 * On success return positive value. On error, negative value.
1128 */
f50f23d9 1129int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
6151a90f 1130 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
d3e2ba59 1131 enum lttng_stream_type type, uint64_t session_id,
fb9a95c4 1132 const char *session_name, const char *hostname,
6fa5fe7c 1133 const char *base_path, int session_live_timer,
46ef2188
MD
1134 const uint64_t *current_chunk_id, time_t session_creation_time,
1135 bool session_name_contains_creation_time)
37278a1e
DG
1136{
1137 int ret;
1138 struct lttcomm_consumer_msg msg;
1139
1140 /* Code flow error. Safety net. */
6151a90f 1141 assert(rsock);
37278a1e 1142 assert(consumer);
f50f23d9 1143 assert(consumer_sock);
37278a1e 1144
53efb85a 1145 memset(&msg, 0, sizeof(msg));
37278a1e
DG
1146 /* Bail out if consumer is disabled */
1147 if (!consumer->enabled) {
f73fabfd 1148 ret = LTTNG_OK;
37278a1e
DG
1149 goto error;
1150 }
1151
d3e2ba59 1152 if (type == LTTNG_STREAM_CONTROL) {
ecd1a12f 1153 char output_path[LTTNG_PATH_MAX] = {};
07aa2e42 1154 uint64_t relayd_session_id;
ecd1a12f 1155
d3e2ba59 1156 ret = relayd_create_session(rsock,
07aa2e42 1157 &relayd_session_id,
6fa5fe7c
MD
1158 session_name, hostname, base_path,
1159 session_live_timer,
658f12fa 1160 consumer->snapshot, session_id,
db1da059 1161 sessiond_uuid, current_chunk_id,
46ef2188 1162 session_creation_time,
ecd1a12f
MD
1163 session_name_contains_creation_time,
1164 output_path);
d3e2ba59
JD
1165 if (ret < 0) {
1166 /* Close the control socket. */
1167 (void) relayd_close(rsock);
1168 goto error;
1169 }
07aa2e42 1170 msg.u.relayd_sock.relayd_session_id = relayd_session_id;
ecd1a12f
MD
1171 DBG("Created session on relay, output path reply: %s",
1172 output_path);
d3e2ba59
JD
1173 }
1174
37278a1e
DG
1175 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1176 /*
1177 * Assign network consumer output index using the temporary consumer since
1178 * this call should only be made from within a set_consumer_uri() function
1179 * call in the session daemon.
1180 */
1181 msg.u.relayd_sock.net_index = consumer->net_seq_index;
1182 msg.u.relayd_sock.type = type;
46e6455f 1183 msg.u.relayd_sock.session_id = session_id;
6151a90f 1184 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
37278a1e 1185
9363801e 1186 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
52898cb1 1187 ret = consumer_send_msg(consumer_sock, &msg);
f50f23d9
DG
1188 if (ret < 0) {
1189 goto error;
1190 }
1191
37278a1e 1192 DBG3("Sending relayd socket file descriptor to consumer");
f712d514 1193 ret = consumer_send_fds(consumer_sock, ALIGNED_CONST_PTR(rsock->sock.fd), 1);
37278a1e
DG
1194 if (ret < 0) {
1195 goto error;
1196 }
1197
1198 DBG2("Consumer relayd socket sent");
1199
1200error:
1201 return ret;
1202}
173af62f 1203
62c43103
JD
1204static
1205int consumer_send_pipe(struct consumer_socket *consumer_sock,
1206 enum lttng_consumer_command cmd, int pipe)
e9404c27
JG
1207{
1208 int ret;
1209 struct lttcomm_consumer_msg msg;
62c43103
JD
1210 const char *pipe_name;
1211 const char *command_name;
1212
1213 switch (cmd) {
1214 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1215 pipe_name = "channel monitor";
1216 command_name = "SET_CHANNEL_MONITOR_PIPE";
1217 break;
62c43103
JD
1218 default:
1219 ERR("Unexpected command received in %s (cmd = %d)", __func__,
1220 (int) cmd);
1221 abort();
1222 }
e9404c27
JG
1223
1224 /* Code flow error. Safety net. */
1225
1226 memset(&msg, 0, sizeof(msg));
62c43103 1227 msg.cmd_type = cmd;
e9404c27 1228
3e4dc117 1229 pthread_mutex_lock(consumer_sock->lock);
62c43103 1230 DBG3("Sending %s command to consumer", command_name);
e9404c27
JG
1231 ret = consumer_send_msg(consumer_sock, &msg);
1232 if (ret < 0) {
1233 goto error;
1234 }
1235
62c43103
JD
1236 DBG3("Sending %s pipe %d to consumer on socket %d",
1237 pipe_name,
e9404c27
JG
1238 pipe, *consumer_sock->fd_ptr);
1239 ret = consumer_send_fds(consumer_sock, &pipe, 1);
1240 if (ret < 0) {
1241 goto error;
1242 }
1243
62c43103 1244 DBG2("%s pipe successfully sent", pipe_name);
e9404c27 1245error:
3e4dc117 1246 pthread_mutex_unlock(consumer_sock->lock);
e9404c27
JG
1247 return ret;
1248}
1249
62c43103
JD
1250int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock,
1251 int pipe)
1252{
1253 return consumer_send_pipe(consumer_sock,
1254 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe);
1255}
1256
806e2684 1257/*
5e280d77
MD
1258 * Ask the consumer if the data is pending for the specific session id.
1259 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
806e2684 1260 */
d88aee68 1261int consumer_is_data_pending(uint64_t session_id,
806e2684
DG
1262 struct consumer_output *consumer)
1263{
1264 int ret;
6d805429 1265 int32_t ret_code = 0; /* Default is that the data is NOT pending */
806e2684
DG
1266 struct consumer_socket *socket;
1267 struct lttng_ht_iter iter;
1268 struct lttcomm_consumer_msg msg;
1269
1270 assert(consumer);
1271
53efb85a 1272 DBG3("Consumer data pending for id %" PRIu64, session_id);
806e2684 1273
53efb85a
MD
1274 memset(&msg, 0, sizeof(msg));
1275 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
d88aee68 1276 msg.u.data_pending.session_id = session_id;
806e2684 1277
c8f59ee5 1278 /* Send command for each consumer */
b82c5c4d 1279 rcu_read_lock();
806e2684
DG
1280 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1281 node.node) {
806e2684 1282 pthread_mutex_lock(socket->lock);
52898cb1 1283 ret = consumer_socket_send(socket, &msg, sizeof(msg));
806e2684 1284 if (ret < 0) {
806e2684 1285 pthread_mutex_unlock(socket->lock);
b82c5c4d 1286 goto error_unlock;
806e2684
DG
1287 }
1288
f50f23d9
DG
1289 /*
1290 * No need for a recv reply status because the answer to the command is
1291 * the reply status message.
1292 */
1293
52898cb1
DG
1294 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1295 if (ret < 0) {
806e2684 1296 pthread_mutex_unlock(socket->lock);
b82c5c4d 1297 goto error_unlock;
806e2684 1298 }
806e2684
DG
1299 pthread_mutex_unlock(socket->lock);
1300
6d805429 1301 if (ret_code == 1) {
806e2684
DG
1302 break;
1303 }
1304 }
b82c5c4d 1305 rcu_read_unlock();
806e2684 1306
d88aee68
DG
1307 DBG("Consumer data is %s pending for session id %" PRIu64,
1308 ret_code == 1 ? "" : "NOT", session_id);
806e2684
DG
1309 return ret_code;
1310
b82c5c4d
DG
1311error_unlock:
1312 rcu_read_unlock();
806e2684
DG
1313 return -1;
1314}
7972aab2
DG
1315
1316/*
1317 * Send a flush command to consumer using the given channel key.
1318 *
1319 * Return 0 on success else a negative value.
1320 */
1321int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1322{
1323 int ret;
1324 struct lttcomm_consumer_msg msg;
1325
1326 assert(socket);
7972aab2
DG
1327
1328 DBG2("Consumer flush channel key %" PRIu64, key);
1329
53efb85a 1330 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1331 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1332 msg.u.flush_channel.key = key;
1333
1334 pthread_mutex_lock(socket->lock);
1335 health_code_update();
1336
1337 ret = consumer_send_msg(socket, &msg);
1338 if (ret < 0) {
1339 goto end;
1340 }
1341
1342end:
1343 health_code_update();
1344 pthread_mutex_unlock(socket->lock);
1345 return ret;
1346}
1347
0dd01979
MD
1348/*
1349 * Send a clear quiescent command to consumer using the given channel key.
1350 *
1351 * Return 0 on success else a negative value.
1352 */
1353int consumer_clear_quiescent_channel(struct consumer_socket *socket, uint64_t key)
1354{
1355 int ret;
1356 struct lttcomm_consumer_msg msg;
1357
1358 assert(socket);
1359
1360 DBG2("Consumer clear quiescent channel key %" PRIu64, key);
1361
1362 memset(&msg, 0, sizeof(msg));
1363 msg.cmd_type = LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL;
1364 msg.u.clear_quiescent_channel.key = key;
1365
1366 pthread_mutex_lock(socket->lock);
1367 health_code_update();
1368
1369 ret = consumer_send_msg(socket, &msg);
1370 if (ret < 0) {
1371 goto end;
1372 }
1373
1374end:
1375 health_code_update();
1376 pthread_mutex_unlock(socket->lock);
1377 return ret;
1378}
1379
7972aab2 1380/*
dc2bbdae
MD
1381 * Send a close metadata command to consumer using the given channel key.
1382 * Called with registry lock held.
7972aab2
DG
1383 *
1384 * Return 0 on success else a negative value.
1385 */
1386int consumer_close_metadata(struct consumer_socket *socket,
1387 uint64_t metadata_key)
1388{
1389 int ret;
1390 struct lttcomm_consumer_msg msg;
1391
1392 assert(socket);
7972aab2
DG
1393
1394 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1395
53efb85a 1396 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1397 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1398 msg.u.close_metadata.key = metadata_key;
1399
1400 pthread_mutex_lock(socket->lock);
1401 health_code_update();
1402
1403 ret = consumer_send_msg(socket, &msg);
1404 if (ret < 0) {
1405 goto end;
1406 }
1407
1408end:
1409 health_code_update();
1410 pthread_mutex_unlock(socket->lock);
1411 return ret;
1412}
1413
1414/*
1415 * Send a setup metdata command to consumer using the given channel key.
1416 *
1417 * Return 0 on success else a negative value.
1418 */
1419int consumer_setup_metadata(struct consumer_socket *socket,
1420 uint64_t metadata_key)
1421{
1422 int ret;
1423 struct lttcomm_consumer_msg msg;
1424
1425 assert(socket);
7972aab2
DG
1426
1427 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1428
53efb85a 1429 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1430 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1431 msg.u.setup_metadata.key = metadata_key;
1432
1433 pthread_mutex_lock(socket->lock);
1434 health_code_update();
1435
1436 ret = consumer_send_msg(socket, &msg);
1437 if (ret < 0) {
1438 goto end;
1439 }
1440
1441end:
1442 health_code_update();
1443 pthread_mutex_unlock(socket->lock);
1444 return ret;
1445}
1446
1447/*
dc2bbdae
MD
1448 * Send metadata string to consumer.
1449 * RCU read-side lock must be held to guarantee existence of socket.
7972aab2
DG
1450 *
1451 * Return 0 on success else a negative value.
1452 */
1453int consumer_push_metadata(struct consumer_socket *socket,
1454 uint64_t metadata_key, char *metadata_str, size_t len,
93ec662e 1455 size_t target_offset, uint64_t version)
7972aab2
DG
1456{
1457 int ret;
1458 struct lttcomm_consumer_msg msg;
1459
1460 assert(socket);
7972aab2 1461
9363801e 1462 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
7972aab2 1463
dc2bbdae
MD
1464 pthread_mutex_lock(socket->lock);
1465
53efb85a 1466 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1467 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1468 msg.u.push_metadata.key = metadata_key;
1469 msg.u.push_metadata.target_offset = target_offset;
1470 msg.u.push_metadata.len = len;
93ec662e 1471 msg.u.push_metadata.version = version;
7972aab2 1472
7972aab2
DG
1473 health_code_update();
1474 ret = consumer_send_msg(socket, &msg);
331744e3 1475 if (ret < 0 || len == 0) {
7972aab2
DG
1476 goto end;
1477 }
1478
9363801e
DG
1479 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1480 len);
7972aab2 1481
52898cb1 1482 ret = consumer_socket_send(socket, metadata_str, len);
7972aab2
DG
1483 if (ret < 0) {
1484 goto end;
1485 }
1486
1487 health_code_update();
1488 ret = consumer_recv_status_reply(socket);
1489 if (ret < 0) {
1490 goto end;
1491 }
1492
1493end:
dc2bbdae 1494 pthread_mutex_unlock(socket->lock);
7972aab2 1495 health_code_update();
7972aab2
DG
1496 return ret;
1497}
6dc3064a
DG
1498
1499/*
1500 * Ask the consumer to snapshot a specific channel using the key.
1501 *
9a654598 1502 * Returns LTTNG_OK on success or else an LTTng error code.
6dc3064a 1503 */
9a654598 1504enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket,
348a81dc 1505 uint64_t key, const struct consumer_output *output, int metadata,
d2956687
JG
1506 uid_t uid, gid_t gid, const char *channel_path, int wait,
1507 uint64_t nb_packets_per_stream)
6dc3064a
DG
1508{
1509 int ret;
9a654598 1510 enum lttng_error_code status = LTTNG_OK;
6dc3064a
DG
1511 struct lttcomm_consumer_msg msg;
1512
1513 assert(socket);
6dc3064a 1514 assert(output);
6dc3064a
DG
1515
1516 DBG("Consumer snapshot channel key %" PRIu64, key);
1517
ee91bab2 1518 memset(&msg, 0, sizeof(msg));
6dc3064a
DG
1519 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1520 msg.u.snapshot_channel.key = key;
d07ceecd 1521 msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
6dc3064a
DG
1522 msg.u.snapshot_channel.metadata = metadata;
1523
348a81dc 1524 if (output->type == CONSUMER_DST_NET) {
d2956687 1525 msg.u.snapshot_channel.relayd_id =
348a81dc 1526 output->net_seq_index;
6dc3064a 1527 msg.u.snapshot_channel.use_relayd = 1;
6dc3064a 1528 } else {
07b86b52 1529 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
d2956687
JG
1530 }
1531 ret = lttng_strncpy(msg.u.snapshot_channel.pathname,
1532 channel_path,
1533 sizeof(msg.u.snapshot_channel.pathname));
1534 if (ret < 0) {
1535 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1536 sizeof(msg.u.snapshot_channel.pathname),
1537 strlen(channel_path),
1538 channel_path);
1539 status = LTTNG_ERR_SNAPSHOT_FAIL;
1540 goto error;
6dc3064a
DG
1541 }
1542
1543 health_code_update();
9d1103e6 1544 pthread_mutex_lock(socket->lock);
6dc3064a 1545 ret = consumer_send_msg(socket, &msg);
9d1103e6 1546 pthread_mutex_unlock(socket->lock);
6dc3064a 1547 if (ret < 0) {
9bbfb88c
MD
1548 switch (-ret) {
1549 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
9a654598 1550 status = LTTNG_ERR_CHAN_NOT_FOUND;
9bbfb88c
MD
1551 break;
1552 default:
9a654598 1553 status = LTTNG_ERR_SNAPSHOT_FAIL;
9bbfb88c
MD
1554 break;
1555 }
6dc3064a
DG
1556 goto error;
1557 }
1558
1559error:
1560 health_code_update();
9a654598 1561 return status;
6dc3064a 1562}
fb83fe64
JD
1563
1564/*
1565 * Ask the consumer the number of discarded events for a channel.
1566 */
1567int consumer_get_discarded_events(uint64_t session_id, uint64_t channel_key,
1568 struct consumer_output *consumer, uint64_t *discarded)
1569{
1570 int ret;
1571 struct consumer_socket *socket;
1572 struct lttng_ht_iter iter;
1573 struct lttcomm_consumer_msg msg;
1574
1575 assert(consumer);
1576
1577 DBG3("Consumer discarded events id %" PRIu64, session_id);
1578
1579 memset(&msg, 0, sizeof(msg));
1580 msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1581 msg.u.discarded_events.session_id = session_id;
1582 msg.u.discarded_events.channel_key = channel_key;
1583
1584 *discarded = 0;
1585
1586 /* Send command for each consumer */
1587 rcu_read_lock();
1588 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1589 node.node) {
1590 uint64_t consumer_discarded = 0;
1591 pthread_mutex_lock(socket->lock);
1592 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1593 if (ret < 0) {
1594 pthread_mutex_unlock(socket->lock);
1595 goto end;
1596 }
1597
1598 /*
1599 * No need for a recv reply status because the answer to the
1600 * command is the reply status message.
1601 */
1602 ret = consumer_socket_recv(socket, &consumer_discarded,
1603 sizeof(consumer_discarded));
1604 if (ret < 0) {
1605 ERR("get discarded events");
1606 pthread_mutex_unlock(socket->lock);
1607 goto end;
1608 }
1609 pthread_mutex_unlock(socket->lock);
1610 *discarded += consumer_discarded;
1611 }
1612 ret = 0;
1613 DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64,
1614 *discarded, session_id);
1615
1616end:
1617 rcu_read_unlock();
1618 return ret;
1619}
1620
1621/*
1622 * Ask the consumer the number of lost packets for a channel.
1623 */
1624int consumer_get_lost_packets(uint64_t session_id, uint64_t channel_key,
1625 struct consumer_output *consumer, uint64_t *lost)
1626{
1627 int ret;
1628 struct consumer_socket *socket;
1629 struct lttng_ht_iter iter;
1630 struct lttcomm_consumer_msg msg;
1631
1632 assert(consumer);
1633
1634 DBG3("Consumer lost packets id %" PRIu64, session_id);
1635
1636 memset(&msg, 0, sizeof(msg));
1637 msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1638 msg.u.lost_packets.session_id = session_id;
1639 msg.u.lost_packets.channel_key = channel_key;
1640
1641 *lost = 0;
1642
1643 /* Send command for each consumer */
1644 rcu_read_lock();
1645 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1646 node.node) {
1647 uint64_t consumer_lost = 0;
1648 pthread_mutex_lock(socket->lock);
1649 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1650 if (ret < 0) {
1651 pthread_mutex_unlock(socket->lock);
1652 goto end;
1653 }
1654
1655 /*
1656 * No need for a recv reply status because the answer to the
1657 * command is the reply status message.
1658 */
1659 ret = consumer_socket_recv(socket, &consumer_lost,
1660 sizeof(consumer_lost));
1661 if (ret < 0) {
1662 ERR("get lost packets");
1663 pthread_mutex_unlock(socket->lock);
1664 goto end;
1665 }
1666 pthread_mutex_unlock(socket->lock);
1667 *lost += consumer_lost;
1668 }
1669 ret = 0;
1670 DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64,
1671 *lost, session_id);
1672
1673end:
1674 rcu_read_unlock();
1675 return ret;
1676}
a1ae2ea5 1677
5c408ad8
JD
1678/*
1679 * Ask the consumer to rotate a channel.
5c408ad8
JD
1680 *
1681 * The new_chunk_id is the session->rotate_count that has been incremented
1682 * when the rotation started. On the relay, this allows to keep track in which
1683 * chunk each stream is currently writing to (for the rotate_pending operation).
1684 */
1685int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key,
1686 uid_t uid, gid_t gid, struct consumer_output *output,
d2956687 1687 bool is_metadata_channel)
5c408ad8
JD
1688{
1689 int ret;
1690 struct lttcomm_consumer_msg msg;
1691
1692 assert(socket);
1693
1694 DBG("Consumer rotate channel key %" PRIu64, key);
1695
1696 pthread_mutex_lock(socket->lock);
1697 memset(&msg, 0, sizeof(msg));
1698 msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL;
1699 msg.u.rotate_channel.key = key;
1700 msg.u.rotate_channel.metadata = !!is_metadata_channel;
5c408ad8
JD
1701
1702 if (output->type == CONSUMER_DST_NET) {
1703 msg.u.rotate_channel.relayd_id = output->net_seq_index;
5c408ad8
JD
1704 } else {
1705 msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL;
5c408ad8
JD
1706 }
1707
1708 health_code_update();
1709 ret = consumer_send_msg(socket, &msg);
1710 if (ret < 0) {
20f37cb4
MD
1711 switch (-ret) {
1712 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1713 ret = -LTTNG_ERR_CHAN_NOT_FOUND;
1714 break;
1715 default:
1716 ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER;
1717 break;
1718 }
5c408ad8
JD
1719 goto error;
1720 }
5c408ad8
JD
1721error:
1722 pthread_mutex_unlock(socket->lock);
1723 health_code_update();
1724 return ret;
1725}
1726
04ed9e10
JG
1727int consumer_open_channel_packets(struct consumer_socket *socket, uint64_t key)
1728{
1729 int ret;
1730 const struct lttcomm_consumer_msg msg = {
1731 .cmd_type = LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS,
1732 .u.open_channel_packets.key = key,
1733 };
1734
1735 assert(socket);
1736
1737 DBG("Consumer open channel packets: channel key = %" PRIu64, key);
1738
1739 health_code_update();
1740
1741 pthread_mutex_lock(socket->lock);
1742 ret = consumer_send_msg(socket, &msg);
1743 pthread_mutex_unlock(socket->lock);
1744 if (ret < 0) {
1745 goto error_socket;
1746 }
1747
1748error_socket:
1749 health_code_update();
1750 return ret;
1751}
1752
51a4828f
MD
1753int consumer_clear_channel(struct consumer_socket *socket, uint64_t key)
1754{
1755 int ret;
1756 struct lttcomm_consumer_msg msg;
1757
1758 assert(socket);
1759
1760 DBG("Consumer clear channel %" PRIu64, key);
1761
1762 memset(&msg, 0, sizeof(msg));
1763 msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL;
1764 msg.u.clear_channel.key = key;
1765
1766 health_code_update();
1767
1768 pthread_mutex_lock(socket->lock);
1769 ret = consumer_send_msg(socket, &msg);
1770 if (ret < 0) {
1771 goto error_socket;
1772 }
1773
1774error_socket:
1775 pthread_mutex_unlock(socket->lock);
1776
1777 health_code_update();
1778 return ret;
1779}
1780
d2956687
JG
1781int consumer_init(struct consumer_socket *socket,
1782 const lttng_uuid sessiond_uuid)
00fb02ac
JD
1783{
1784 int ret;
d2956687
JG
1785 struct lttcomm_consumer_msg msg = {
1786 .cmd_type = LTTNG_CONSUMER_INIT,
1787 };
00fb02ac
JD
1788
1789 assert(socket);
00fb02ac 1790
d2956687
JG
1791 DBG("Sending consumer initialization command");
1792 lttng_uuid_copy(msg.u.init.sessiond_uuid, sessiond_uuid);
00fb02ac
JD
1793
1794 health_code_update();
1795 ret = consumer_send_msg(socket, &msg);
1796 if (ret < 0) {
1797 goto error;
1798 }
1799
d88744a4
JD
1800error:
1801 health_code_update();
1802 return ret;
1803}
1804
1805/*
d2956687 1806 * Ask the consumer to create a new chunk for a given session.
92816cc3 1807 *
d2956687 1808 * Called with the consumer socket lock held.
92816cc3 1809 */
d2956687
JG
1810int consumer_create_trace_chunk(struct consumer_socket *socket,
1811 uint64_t relayd_id, uint64_t session_id,
5da88b0f
MD
1812 struct lttng_trace_chunk *chunk,
1813 const char *domain_subdir)
92816cc3
JG
1814{
1815 int ret;
d2956687
JG
1816 enum lttng_trace_chunk_status chunk_status;
1817 struct lttng_credentials chunk_credentials;
5da88b0f
MD
1818 const struct lttng_directory_handle *chunk_directory_handle = NULL;
1819 struct lttng_directory_handle *domain_handle = NULL;
1820 int domain_dirfd;
d2956687 1821 const char *chunk_name;
913a542b 1822 bool chunk_name_overridden;
d2956687
JG
1823 uint64_t chunk_id;
1824 time_t creation_timestamp;
1825 char creation_timestamp_buffer[ISO8601_STR_LEN];
1826 const char *creation_timestamp_str = "(none)";
1827 const bool chunk_has_local_output = relayd_id == -1ULL;
69ebf37e 1828 enum lttng_trace_chunk_status tc_status;
d2956687
JG
1829 struct lttcomm_consumer_msg msg = {
1830 .cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
1831 .u.create_trace_chunk.session_id = session_id,
1832 };
92816cc3
JG
1833
1834 assert(socket);
d2956687 1835 assert(chunk);
92816cc3 1836
d2956687
JG
1837 if (relayd_id != -1ULL) {
1838 LTTNG_OPTIONAL_SET(&msg.u.create_trace_chunk.relayd_id,
1839 relayd_id);
1840 }
92816cc3 1841
d2956687 1842 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name,
913a542b 1843 &chunk_name_overridden);
d2956687
JG
1844 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1845 chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1846 ERR("Failed to get name of trace chunk");
1847 ret = -LTTNG_ERR_FATAL;
92816cc3
JG
1848 goto error;
1849 }
913a542b 1850 if (chunk_name_overridden) {
d2956687
JG
1851 ret = lttng_strncpy(msg.u.create_trace_chunk.override_name,
1852 chunk_name,
1853 sizeof(msg.u.create_trace_chunk.override_name));
1854 if (ret) {
1855 ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1856 chunk_name);
1857 ret = -LTTNG_ERR_FATAL;
1858 goto error;
1859 }
1860 }
92816cc3 1861
d2956687
JG
1862 chunk_status = lttng_trace_chunk_get_creation_timestamp(chunk,
1863 &creation_timestamp);
1864 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1865 ret = -LTTNG_ERR_FATAL;
92816cc3
JG
1866 goto error;
1867 }
d2956687
JG
1868 msg.u.create_trace_chunk.creation_timestamp =
1869 (uint64_t) creation_timestamp;
1870 /* Only used for logging purposes. */
1871 ret = time_to_iso8601_str(creation_timestamp,
1872 creation_timestamp_buffer,
1873 sizeof(creation_timestamp_buffer));
1874 creation_timestamp_str = !ret ? creation_timestamp_buffer :
1875 "(formatting error)";
1876
1877 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1878 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1879 /*
1880 * Anonymous trace chunks should never be transmitted
1881 * to remote peers (consumerd and relayd). They are used
1882 * internally for backward-compatibility purposes.
1883 */
1884 ret = -LTTNG_ERR_FATAL;
1885 goto error;
1886 }
1887 msg.u.create_trace_chunk.chunk_id = chunk_id;
92816cc3 1888
d2956687 1889 if (chunk_has_local_output) {
cbf53d23 1890 chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle(
d2956687
JG
1891 chunk, &chunk_directory_handle);
1892 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1893 ret = -LTTNG_ERR_FATAL;
1894 goto error;
1895 }
e5add6d0
JG
1896 chunk_status = lttng_trace_chunk_get_credentials(
1897 chunk, &chunk_credentials);
1898 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1899 /*
1900 * Not associating credentials to a sessiond chunk is a
1901 * fatal internal error.
1902 */
1903 ret = -LTTNG_ERR_FATAL;
1904 goto error;
1905 }
69ebf37e
JR
1906 tc_status = lttng_trace_chunk_create_subdirectory(
1907 chunk, domain_subdir);
1908 if (tc_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
5da88b0f
MD
1909 PERROR("Failed to create chunk domain output directory \"%s\"",
1910 domain_subdir);
1911 ret = -LTTNG_ERR_FATAL;
1912 goto error;
1913 }
1914 domain_handle = lttng_directory_handle_create_from_handle(
1915 domain_subdir,
1916 chunk_directory_handle);
1917 if (!domain_handle) {
1918 ret = -LTTNG_ERR_FATAL;
1919 goto error;
1920 }
1921
1922 /*
1923 * This will only compile on platforms that support
1924 * dirfd (POSIX.2008). This is fine as the session daemon
1925 * is only built for such platforms.
1926 *
1927 * The ownership of the chunk directory handle's is maintained
1928 * by the trace chunk.
1929 */
1930 domain_dirfd = lttng_directory_handle_get_dirfd(
1931 domain_handle);
1932 assert(domain_dirfd >= 0);
1933
e5add6d0 1934 msg.u.create_trace_chunk.credentials.value.uid =
2463b787 1935 lttng_credentials_get_uid(&chunk_credentials);
e5add6d0 1936 msg.u.create_trace_chunk.credentials.value.gid =
2463b787 1937 lttng_credentials_get_gid(&chunk_credentials);
e5add6d0 1938 msg.u.create_trace_chunk.credentials.is_set = 1;
d2956687 1939 }
d88744a4 1940
d2956687
JG
1941 DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1942 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
1943 ", creation_timestamp = %s",
1944 relayd_id, session_id, chunk_id,
1945 creation_timestamp_str);
d88744a4
JD
1946 health_code_update();
1947 ret = consumer_send_msg(socket, &msg);
d2956687 1948 health_code_update();
d88744a4 1949 if (ret < 0) {
d2956687
JG
1950 ERR("Trace chunk creation error on consumer");
1951 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
d88744a4
JD
1952 goto error;
1953 }
1954
d2956687 1955 if (chunk_has_local_output) {
5da88b0f 1956 DBG("Sending trace chunk domain directory fd to consumer");
d2956687 1957 health_code_update();
5da88b0f 1958 ret = consumer_send_fds(socket, &domain_dirfd, 1);
d2956687
JG
1959 health_code_update();
1960 if (ret < 0) {
1961 ERR("Trace chunk creation error on consumer");
1962 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1963 goto error;
1964 }
d88744a4 1965 }
00fb02ac 1966error:
5da88b0f 1967 lttng_directory_handle_put(domain_handle);
00fb02ac
JD
1968 return ret;
1969}
1970
a1ae2ea5 1971/*
d2956687 1972 * Ask the consumer to close a trace chunk for a given session.
a1ae2ea5
JD
1973 *
1974 * Called with the consumer socket lock held.
1975 */
d2956687
JG
1976int consumer_close_trace_chunk(struct consumer_socket *socket,
1977 uint64_t relayd_id, uint64_t session_id,
ecd1a12f
MD
1978 struct lttng_trace_chunk *chunk,
1979 char *closed_trace_chunk_path)
a1ae2ea5
JD
1980{
1981 int ret;
d2956687
JG
1982 enum lttng_trace_chunk_status chunk_status;
1983 struct lttcomm_consumer_msg msg = {
bbc4768c
JG
1984 .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
1985 .u.close_trace_chunk.session_id = session_id,
d2956687 1986 };
ecd1a12f 1987 struct lttcomm_consumer_close_trace_chunk_reply reply;
d2956687
JG
1988 uint64_t chunk_id;
1989 time_t close_timestamp;
bbc4768c
JG
1990 enum lttng_trace_chunk_command_type close_command;
1991 const char *close_command_name = "none";
ecd1a12f 1992 struct lttng_dynamic_buffer path_reception_buffer;
a1ae2ea5
JD
1993
1994 assert(socket);
ecd1a12f 1995 lttng_dynamic_buffer_init(&path_reception_buffer);
a1ae2ea5 1996
d2956687 1997 if (relayd_id != -1ULL) {
bbc4768c
JG
1998 LTTNG_OPTIONAL_SET(
1999 &msg.u.close_trace_chunk.relayd_id, relayd_id);
2000 }
2001
2002 chunk_status = lttng_trace_chunk_get_close_command(
2003 chunk, &close_command);
2004 switch (chunk_status) {
2005 case LTTNG_TRACE_CHUNK_STATUS_OK:
2006 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command,
2007 (uint32_t) close_command);
2008 break;
2009 case LTTNG_TRACE_CHUNK_STATUS_NONE:
2010 break;
2011 default:
2012 ERR("Failed to get trace chunk close command");
2013 ret = -1;
2014 goto error;
a1ae2ea5
JD
2015 }
2016
d2956687
JG
2017 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2018 /*
2019 * Anonymous trace chunks should never be transmitted to remote peers
2020 * (consumerd and relayd). They are used internally for
2021 * backward-compatibility purposes.
2022 */
2023 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2024 msg.u.close_trace_chunk.chunk_id = chunk_id;
2025
2026 chunk_status = lttng_trace_chunk_get_close_timestamp(chunk,
2027 &close_timestamp);
2028 /*
2029 * A trace chunk should be closed locally before being closed remotely.
2030 * Otherwise, the close timestamp would never be transmitted to the
2031 * peers.
2032 */
2033 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2034 msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp;
2035
bbc4768c
JG
2036 if (msg.u.close_trace_chunk.close_command.is_set) {
2037 close_command_name = lttng_trace_chunk_command_type_get_name(
2038 close_command);
2039 }
d2956687 2040 DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
bbc4768c
JG
2041 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
2042 ", close command = \"%s\"",
2043 relayd_id, session_id, chunk_id, close_command_name);
a1ae2ea5
JD
2044
2045 health_code_update();
ecd1a12f 2046 ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg));
a1ae2ea5 2047 if (ret < 0) {
d2956687 2048 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
a1ae2ea5
JD
2049 goto error;
2050 }
ecd1a12f
MD
2051 ret = consumer_socket_recv(socket, &reply, sizeof(reply));
2052 if (ret < 0) {
2053 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2054 goto error;
2055 }
2056 if (reply.path_length >= LTTNG_PATH_MAX) {
2057 ERR("Invalid path returned by relay daemon: %" PRIu32 "bytes exceeds maximal allowed length of %d bytes",
2058 reply.path_length, LTTNG_PATH_MAX);
2059 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2060 goto error;
2061 }
2062 ret = lttng_dynamic_buffer_set_size(&path_reception_buffer,
2063 reply.path_length);
2064 if (ret) {
2065 ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command");
2066 ret = -LTTNG_ERR_NOMEM;
2067 goto error;
2068 }
2069 ret = consumer_socket_recv(socket, path_reception_buffer.data,
2070 path_reception_buffer.size);
2071 if (ret < 0) {
2072 ERR("Communication error while receiving path of closed trace chunk");
2073 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2074 goto error;
2075 }
2076 if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') {
2077 ERR("Invalid path returned by relay daemon: not null-terminated");
2078 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2079 goto error;
2080 }
2081 if (closed_trace_chunk_path) {
2082 /*
2083 * closed_trace_chunk_path is assumed to have a length >=
2084 * LTTNG_PATH_MAX
2085 */
2086 memcpy(closed_trace_chunk_path, path_reception_buffer.data,
2087 path_reception_buffer.size);
2088 }
a1ae2ea5 2089error:
ecd1a12f 2090 lttng_dynamic_buffer_reset(&path_reception_buffer);
a1ae2ea5
JD
2091 health_code_update();
2092 return ret;
2093}
3654ed19 2094
d2956687
JG
2095/*
2096 * Ask the consumer if a trace chunk exists.
2097 *
2098 * Called with the consumer socket lock held.
2099 * Returns 0 on success, or a negative value on error.
2100 */
2101int consumer_trace_chunk_exists(struct consumer_socket *socket,
2102 uint64_t relayd_id, uint64_t session_id,
2103 struct lttng_trace_chunk *chunk,
2104 enum consumer_trace_chunk_exists_status *result)
3654ed19
JG
2105{
2106 int ret;
d2956687 2107 enum lttng_trace_chunk_status chunk_status;
3654ed19 2108 struct lttcomm_consumer_msg msg = {
d2956687
JG
2109 .cmd_type = LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,
2110 .u.trace_chunk_exists.session_id = session_id,
3654ed19 2111 };
d2956687
JG
2112 uint64_t chunk_id;
2113 const char *consumer_reply_str;
3654ed19
JG
2114
2115 assert(socket);
2116
d2956687
JG
2117 if (relayd_id != -1ULL) {
2118 LTTNG_OPTIONAL_SET(&msg.u.trace_chunk_exists.relayd_id,
2119 relayd_id);
2120 }
2121
2122 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2123 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2124 /*
2125 * Anonymous trace chunks should never be transmitted
2126 * to remote peers (consumerd and relayd). They are used
2127 * internally for backward-compatibility purposes.
2128 */
2129 ret = -LTTNG_ERR_FATAL;
2130 goto error;
2131 }
2132 msg.u.trace_chunk_exists.chunk_id = chunk_id;
2133
2134 DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
2135 ", session_id = %" PRIu64
2136 ", chunk_id = %" PRIu64, relayd_id, session_id, chunk_id);
3654ed19
JG
2137
2138 health_code_update();
2139 ret = consumer_send_msg(socket, &msg);
d2956687
JG
2140 switch (-ret) {
2141 case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK:
2142 consumer_reply_str = "unknown trace chunk";
2143 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK;
2144 break;
2145 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL:
2146 consumer_reply_str = "trace chunk exists locally";
2147 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL;
2148 break;
2149 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE:
2150 consumer_reply_str = "trace chunk exists on remote peer";
2151 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE;
2152 break;
2153 default:
2154 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
2155 ret = -1;
3654ed19
JG
2156 goto error;
2157 }
d2956687
JG
2158 DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
2159 consumer_reply_str);
2160 ret = 0;
3654ed19
JG
2161error:
2162 health_code_update();
2163 return ret;
2164}
This page took 0.30768 seconds and 5 git commands to generate.