Clean-up: sessiond consumer: change space to tabs
[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 }
1b0bebcb
JG
954 msg->u.ask_channel.buffer_credentials.uid = buffer_credentials->uid;
955 msg->u.ask_channel.buffer_credentials.gid = buffer_credentials->gid;
ffe60014
DG
956
957 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
958 msg->u.ask_channel.subbuf_size = subbuf_size;
959 msg->u.ask_channel.num_subbuf = num_subbuf ;
960 msg->u.ask_channel.overwrite = overwrite;
961 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
962 msg->u.ask_channel.read_timer_interval = read_timer_interval;
ecc48a90 963 msg->u.ask_channel.live_timer_interval = live_timer_interval;
a2814ea7 964 msg->u.ask_channel.is_live = is_in_live_session;
e9404c27 965 msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
ffe60014
DG
966 msg->u.ask_channel.output = output;
967 msg->u.ask_channel.type = type;
968 msg->u.ask_channel.session_id = session_id;
1950109e 969 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
ffe60014
DG
970 msg->u.ask_channel.relayd_id = relayd_id;
971 msg->u.ask_channel.key = key;
7972aab2 972 msg->u.ask_channel.chan_id = chan_id;
1624d5b7
JD
973 msg->u.ask_channel.tracefile_size = tracefile_size;
974 msg->u.ask_channel.tracefile_count = tracefile_count;
2bba9e53 975 msg->u.ask_channel.monitor = monitor;
567eb353 976 msg->u.ask_channel.ust_app_uid = ust_app_uid;
491d1539 977 msg->u.ask_channel.blocking_timeout = blocking_timeout;
ffe60014
DG
978
979 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
980
10a50311
JD
981 if (pathname) {
982 strncpy(msg->u.ask_channel.pathname, pathname,
983 sizeof(msg->u.ask_channel.pathname));
984 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
985 }
ffe60014
DG
986
987 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
988 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
d7ba1388 989
3d071855
MD
990 if (root_shm_path) {
991 strncpy(msg->u.ask_channel.root_shm_path, root_shm_path,
992 sizeof(msg->u.ask_channel.root_shm_path));
993 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] = '\0';
994 }
d7ba1388
MD
995 if (shm_path) {
996 strncpy(msg->u.ask_channel.shm_path, shm_path,
997 sizeof(msg->u.ask_channel.shm_path));
998 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
999 }
ffe60014
DG
1000}
1001
00e2e675
DG
1002/*
1003 * Init channel communication message structure.
1004 */
638e7b4e 1005void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg,
d88aee68 1006 uint64_t channel_key,
ffe60014
DG
1007 uint64_t session_id,
1008 const char *pathname,
1009 uid_t uid,
1010 gid_t gid,
d88aee68 1011 uint64_t relayd_id,
c30aaa51 1012 const char *name,
ffe60014
DG
1013 unsigned int nb_init_streams,
1014 enum lttng_event_output output,
1624d5b7
JD
1015 int type,
1016 uint64_t tracefile_size,
2bba9e53 1017 uint64_t tracefile_count,
ecc48a90 1018 unsigned int monitor,
e9404c27 1019 unsigned int live_timer_interval,
a2814ea7 1020 bool is_in_live_session,
d2956687
JG
1021 unsigned int monitor_timer_interval,
1022 struct lttng_trace_chunk *trace_chunk)
00e2e675
DG
1023{
1024 assert(msg);
1025
00e2e675
DG
1026 /* Zeroed structure */
1027 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1028
26c468bb 1029 if (trace_chunk) {
d2956687
JG
1030 uint64_t chunk_id;
1031 enum lttng_trace_chunk_status chunk_status;
1032
1033 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
1034 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
1035 LTTNG_OPTIONAL_SET(&msg->u.channel.chunk_id, chunk_id);
26c468bb 1036 }
d2956687 1037
00e2e675 1038 /* Send channel */
638e7b4e 1039 msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
00e2e675 1040 msg->u.channel.channel_key = channel_key;
ffe60014 1041 msg->u.channel.session_id = session_id;
ffe60014 1042 msg->u.channel.relayd_id = relayd_id;
c30aaa51 1043 msg->u.channel.nb_init_streams = nb_init_streams;
ffe60014
DG
1044 msg->u.channel.output = output;
1045 msg->u.channel.type = type;
1624d5b7
JD
1046 msg->u.channel.tracefile_size = tracefile_size;
1047 msg->u.channel.tracefile_count = tracefile_count;
2bba9e53 1048 msg->u.channel.monitor = monitor;
ecc48a90 1049 msg->u.channel.live_timer_interval = live_timer_interval;
a2814ea7 1050 msg->u.channel.is_live = is_in_live_session;
e9404c27 1051 msg->u.channel.monitor_timer_interval = monitor_timer_interval;
ffe60014
DG
1052
1053 strncpy(msg->u.channel.pathname, pathname,
1054 sizeof(msg->u.channel.pathname));
1055 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
1056
1057 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
1058 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
00e2e675
DG
1059}
1060
1061/*
1062 * Init stream communication message structure.
1063 */
e098433c 1064void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg,
d88aee68
DG
1065 uint64_t channel_key,
1066 uint64_t stream_key,
d2956687 1067 int32_t cpu)
00e2e675
DG
1068{
1069 assert(msg);
1070
1071 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1072
e098433c 1073 msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM;
00e2e675
DG
1074 msg->u.stream.channel_key = channel_key;
1075 msg->u.stream.stream_key = stream_key;
ffe60014 1076 msg->u.stream.cpu = cpu;
00e2e675
DG
1077}
1078
a4baae1b
JD
1079void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
1080 enum lttng_consumer_command cmd,
1081 uint64_t channel_key, uint64_t net_seq_idx)
1082{
1083 assert(msg);
1084
1085 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1086
1087 msg->cmd_type = cmd;
1088 msg->u.sent_streams.channel_key = channel_key;
1089 msg->u.sent_streams.net_seq_idx = net_seq_idx;
1090}
1091
00e2e675
DG
1092/*
1093 * Send stream communication structure to the consumer.
1094 */
f50f23d9
DG
1095int consumer_send_stream(struct consumer_socket *sock,
1096 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
ac2f30af 1097 const int *fds, size_t nb_fd)
00e2e675
DG
1098{
1099 int ret;
1100
1101 assert(msg);
1102 assert(dst);
f50f23d9 1103 assert(sock);
ffe60014 1104 assert(fds);
00e2e675 1105
52898cb1 1106 ret = consumer_send_msg(sock, msg);
f50f23d9
DG
1107 if (ret < 0) {
1108 goto error;
1109 }
1110
00e2e675
DG
1111 ret = consumer_send_fds(sock, fds, nb_fd);
1112 if (ret < 0) {
1113 goto error;
1114 }
1115
1116error:
1117 return ret;
1118}
37278a1e
DG
1119
1120/*
1121 * Send relayd socket to consumer associated with a session name.
1122 *
43fade62
JG
1123 * The consumer socket lock must be held by the caller.
1124 *
37278a1e
DG
1125 * On success return positive value. On error, negative value.
1126 */
f50f23d9 1127int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
6151a90f 1128 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
d3e2ba59 1129 enum lttng_stream_type type, uint64_t session_id,
fb9a95c4 1130 const char *session_name, const char *hostname,
6fa5fe7c 1131 const char *base_path, int session_live_timer,
46ef2188
MD
1132 const uint64_t *current_chunk_id, time_t session_creation_time,
1133 bool session_name_contains_creation_time)
37278a1e
DG
1134{
1135 int ret;
1136 struct lttcomm_consumer_msg msg;
1137
1138 /* Code flow error. Safety net. */
6151a90f 1139 assert(rsock);
37278a1e 1140 assert(consumer);
f50f23d9 1141 assert(consumer_sock);
37278a1e 1142
53efb85a 1143 memset(&msg, 0, sizeof(msg));
37278a1e
DG
1144 /* Bail out if consumer is disabled */
1145 if (!consumer->enabled) {
f73fabfd 1146 ret = LTTNG_OK;
37278a1e
DG
1147 goto error;
1148 }
1149
d3e2ba59 1150 if (type == LTTNG_STREAM_CONTROL) {
ecd1a12f 1151 char output_path[LTTNG_PATH_MAX] = {};
07aa2e42 1152 uint64_t relayd_session_id;
ecd1a12f 1153
d3e2ba59 1154 ret = relayd_create_session(rsock,
07aa2e42 1155 &relayd_session_id,
6fa5fe7c
MD
1156 session_name, hostname, base_path,
1157 session_live_timer,
658f12fa 1158 consumer->snapshot, session_id,
db1da059 1159 sessiond_uuid, current_chunk_id,
46ef2188 1160 session_creation_time,
ecd1a12f
MD
1161 session_name_contains_creation_time,
1162 output_path);
d3e2ba59
JD
1163 if (ret < 0) {
1164 /* Close the control socket. */
1165 (void) relayd_close(rsock);
1166 goto error;
1167 }
07aa2e42 1168 msg.u.relayd_sock.relayd_session_id = relayd_session_id;
ecd1a12f
MD
1169 DBG("Created session on relay, output path reply: %s",
1170 output_path);
d3e2ba59
JD
1171 }
1172
37278a1e
DG
1173 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1174 /*
1175 * Assign network consumer output index using the temporary consumer since
1176 * this call should only be made from within a set_consumer_uri() function
1177 * call in the session daemon.
1178 */
1179 msg.u.relayd_sock.net_index = consumer->net_seq_index;
1180 msg.u.relayd_sock.type = type;
46e6455f 1181 msg.u.relayd_sock.session_id = session_id;
6151a90f 1182 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
37278a1e 1183
9363801e 1184 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
52898cb1 1185 ret = consumer_send_msg(consumer_sock, &msg);
f50f23d9
DG
1186 if (ret < 0) {
1187 goto error;
1188 }
1189
37278a1e 1190 DBG3("Sending relayd socket file descriptor to consumer");
f712d514 1191 ret = consumer_send_fds(consumer_sock, ALIGNED_CONST_PTR(rsock->sock.fd), 1);
37278a1e
DG
1192 if (ret < 0) {
1193 goto error;
1194 }
1195
1196 DBG2("Consumer relayd socket sent");
1197
1198error:
1199 return ret;
1200}
173af62f 1201
62c43103
JD
1202static
1203int consumer_send_pipe(struct consumer_socket *consumer_sock,
1204 enum lttng_consumer_command cmd, int pipe)
e9404c27
JG
1205{
1206 int ret;
1207 struct lttcomm_consumer_msg msg;
62c43103
JD
1208 const char *pipe_name;
1209 const char *command_name;
1210
1211 switch (cmd) {
1212 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1213 pipe_name = "channel monitor";
1214 command_name = "SET_CHANNEL_MONITOR_PIPE";
1215 break;
62c43103
JD
1216 default:
1217 ERR("Unexpected command received in %s (cmd = %d)", __func__,
1218 (int) cmd);
1219 abort();
1220 }
e9404c27
JG
1221
1222 /* Code flow error. Safety net. */
1223
1224 memset(&msg, 0, sizeof(msg));
62c43103 1225 msg.cmd_type = cmd;
e9404c27 1226
3e4dc117 1227 pthread_mutex_lock(consumer_sock->lock);
62c43103 1228 DBG3("Sending %s command to consumer", command_name);
e9404c27
JG
1229 ret = consumer_send_msg(consumer_sock, &msg);
1230 if (ret < 0) {
1231 goto error;
1232 }
1233
62c43103
JD
1234 DBG3("Sending %s pipe %d to consumer on socket %d",
1235 pipe_name,
e9404c27
JG
1236 pipe, *consumer_sock->fd_ptr);
1237 ret = consumer_send_fds(consumer_sock, &pipe, 1);
1238 if (ret < 0) {
1239 goto error;
1240 }
1241
62c43103 1242 DBG2("%s pipe successfully sent", pipe_name);
e9404c27 1243error:
3e4dc117 1244 pthread_mutex_unlock(consumer_sock->lock);
e9404c27
JG
1245 return ret;
1246}
1247
62c43103
JD
1248int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock,
1249 int pipe)
1250{
1251 return consumer_send_pipe(consumer_sock,
1252 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe);
1253}
1254
806e2684 1255/*
5e280d77
MD
1256 * Ask the consumer if the data is pending for the specific session id.
1257 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
806e2684 1258 */
d88aee68 1259int consumer_is_data_pending(uint64_t session_id,
806e2684
DG
1260 struct consumer_output *consumer)
1261{
1262 int ret;
6d805429 1263 int32_t ret_code = 0; /* Default is that the data is NOT pending */
806e2684
DG
1264 struct consumer_socket *socket;
1265 struct lttng_ht_iter iter;
1266 struct lttcomm_consumer_msg msg;
1267
1268 assert(consumer);
1269
53efb85a 1270 DBG3("Consumer data pending for id %" PRIu64, session_id);
806e2684 1271
53efb85a
MD
1272 memset(&msg, 0, sizeof(msg));
1273 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
d88aee68 1274 msg.u.data_pending.session_id = session_id;
806e2684 1275
c8f59ee5 1276 /* Send command for each consumer */
b82c5c4d 1277 rcu_read_lock();
806e2684
DG
1278 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1279 node.node) {
806e2684 1280 pthread_mutex_lock(socket->lock);
52898cb1 1281 ret = consumer_socket_send(socket, &msg, sizeof(msg));
806e2684 1282 if (ret < 0) {
806e2684 1283 pthread_mutex_unlock(socket->lock);
b82c5c4d 1284 goto error_unlock;
806e2684
DG
1285 }
1286
f50f23d9
DG
1287 /*
1288 * No need for a recv reply status because the answer to the command is
1289 * the reply status message.
1290 */
1291
52898cb1
DG
1292 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1293 if (ret < 0) {
806e2684 1294 pthread_mutex_unlock(socket->lock);
b82c5c4d 1295 goto error_unlock;
806e2684 1296 }
806e2684
DG
1297 pthread_mutex_unlock(socket->lock);
1298
6d805429 1299 if (ret_code == 1) {
806e2684
DG
1300 break;
1301 }
1302 }
b82c5c4d 1303 rcu_read_unlock();
806e2684 1304
d88aee68
DG
1305 DBG("Consumer data is %s pending for session id %" PRIu64,
1306 ret_code == 1 ? "" : "NOT", session_id);
806e2684
DG
1307 return ret_code;
1308
b82c5c4d
DG
1309error_unlock:
1310 rcu_read_unlock();
806e2684
DG
1311 return -1;
1312}
7972aab2
DG
1313
1314/*
1315 * Send a flush command to consumer using the given channel key.
1316 *
1317 * Return 0 on success else a negative value.
1318 */
1319int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1320{
1321 int ret;
1322 struct lttcomm_consumer_msg msg;
1323
1324 assert(socket);
7972aab2
DG
1325
1326 DBG2("Consumer flush channel key %" PRIu64, key);
1327
53efb85a 1328 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1329 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1330 msg.u.flush_channel.key = key;
1331
1332 pthread_mutex_lock(socket->lock);
1333 health_code_update();
1334
1335 ret = consumer_send_msg(socket, &msg);
1336 if (ret < 0) {
1337 goto end;
1338 }
1339
1340end:
1341 health_code_update();
1342 pthread_mutex_unlock(socket->lock);
1343 return ret;
1344}
1345
0dd01979
MD
1346/*
1347 * Send a clear quiescent command to consumer using the given channel key.
1348 *
1349 * Return 0 on success else a negative value.
1350 */
1351int consumer_clear_quiescent_channel(struct consumer_socket *socket, uint64_t key)
1352{
1353 int ret;
1354 struct lttcomm_consumer_msg msg;
1355
1356 assert(socket);
1357
1358 DBG2("Consumer clear quiescent channel key %" PRIu64, key);
1359
1360 memset(&msg, 0, sizeof(msg));
1361 msg.cmd_type = LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL;
1362 msg.u.clear_quiescent_channel.key = key;
1363
1364 pthread_mutex_lock(socket->lock);
1365 health_code_update();
1366
1367 ret = consumer_send_msg(socket, &msg);
1368 if (ret < 0) {
1369 goto end;
1370 }
1371
1372end:
1373 health_code_update();
1374 pthread_mutex_unlock(socket->lock);
1375 return ret;
1376}
1377
7972aab2 1378/*
dc2bbdae
MD
1379 * Send a close metadata command to consumer using the given channel key.
1380 * Called with registry lock held.
7972aab2
DG
1381 *
1382 * Return 0 on success else a negative value.
1383 */
1384int consumer_close_metadata(struct consumer_socket *socket,
1385 uint64_t metadata_key)
1386{
1387 int ret;
1388 struct lttcomm_consumer_msg msg;
1389
1390 assert(socket);
7972aab2
DG
1391
1392 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1393
53efb85a 1394 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1395 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1396 msg.u.close_metadata.key = metadata_key;
1397
1398 pthread_mutex_lock(socket->lock);
1399 health_code_update();
1400
1401 ret = consumer_send_msg(socket, &msg);
1402 if (ret < 0) {
1403 goto end;
1404 }
1405
1406end:
1407 health_code_update();
1408 pthread_mutex_unlock(socket->lock);
1409 return ret;
1410}
1411
1412/*
1413 * Send a setup metdata command to consumer using the given channel key.
1414 *
1415 * Return 0 on success else a negative value.
1416 */
1417int consumer_setup_metadata(struct consumer_socket *socket,
1418 uint64_t metadata_key)
1419{
1420 int ret;
1421 struct lttcomm_consumer_msg msg;
1422
1423 assert(socket);
7972aab2
DG
1424
1425 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1426
53efb85a 1427 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1428 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1429 msg.u.setup_metadata.key = metadata_key;
1430
1431 pthread_mutex_lock(socket->lock);
1432 health_code_update();
1433
1434 ret = consumer_send_msg(socket, &msg);
1435 if (ret < 0) {
1436 goto end;
1437 }
1438
1439end:
1440 health_code_update();
1441 pthread_mutex_unlock(socket->lock);
1442 return ret;
1443}
1444
1445/*
dc2bbdae
MD
1446 * Send metadata string to consumer.
1447 * RCU read-side lock must be held to guarantee existence of socket.
7972aab2
DG
1448 *
1449 * Return 0 on success else a negative value.
1450 */
1451int consumer_push_metadata(struct consumer_socket *socket,
1452 uint64_t metadata_key, char *metadata_str, size_t len,
93ec662e 1453 size_t target_offset, uint64_t version)
7972aab2
DG
1454{
1455 int ret;
1456 struct lttcomm_consumer_msg msg;
1457
1458 assert(socket);
7972aab2 1459
9363801e 1460 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
7972aab2 1461
dc2bbdae
MD
1462 pthread_mutex_lock(socket->lock);
1463
53efb85a 1464 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1465 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1466 msg.u.push_metadata.key = metadata_key;
1467 msg.u.push_metadata.target_offset = target_offset;
1468 msg.u.push_metadata.len = len;
93ec662e 1469 msg.u.push_metadata.version = version;
7972aab2 1470
7972aab2
DG
1471 health_code_update();
1472 ret = consumer_send_msg(socket, &msg);
331744e3 1473 if (ret < 0 || len == 0) {
7972aab2
DG
1474 goto end;
1475 }
1476
9363801e
DG
1477 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1478 len);
7972aab2 1479
52898cb1 1480 ret = consumer_socket_send(socket, metadata_str, len);
7972aab2
DG
1481 if (ret < 0) {
1482 goto end;
1483 }
1484
1485 health_code_update();
1486 ret = consumer_recv_status_reply(socket);
1487 if (ret < 0) {
1488 goto end;
1489 }
1490
1491end:
dc2bbdae 1492 pthread_mutex_unlock(socket->lock);
7972aab2 1493 health_code_update();
7972aab2
DG
1494 return ret;
1495}
6dc3064a
DG
1496
1497/*
1498 * Ask the consumer to snapshot a specific channel using the key.
1499 *
9a654598 1500 * Returns LTTNG_OK on success or else an LTTng error code.
6dc3064a 1501 */
9a654598 1502enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket,
348a81dc 1503 uint64_t key, const struct consumer_output *output, int metadata,
d2956687
JG
1504 uid_t uid, gid_t gid, const char *channel_path, int wait,
1505 uint64_t nb_packets_per_stream)
6dc3064a
DG
1506{
1507 int ret;
9a654598 1508 enum lttng_error_code status = LTTNG_OK;
6dc3064a
DG
1509 struct lttcomm_consumer_msg msg;
1510
1511 assert(socket);
6dc3064a 1512 assert(output);
6dc3064a
DG
1513
1514 DBG("Consumer snapshot channel key %" PRIu64, key);
1515
ee91bab2 1516 memset(&msg, 0, sizeof(msg));
6dc3064a
DG
1517 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1518 msg.u.snapshot_channel.key = key;
d07ceecd 1519 msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
6dc3064a
DG
1520 msg.u.snapshot_channel.metadata = metadata;
1521
348a81dc 1522 if (output->type == CONSUMER_DST_NET) {
d2956687 1523 msg.u.snapshot_channel.relayd_id =
348a81dc 1524 output->net_seq_index;
6dc3064a 1525 msg.u.snapshot_channel.use_relayd = 1;
6dc3064a 1526 } else {
07b86b52 1527 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
d2956687
JG
1528 }
1529 ret = lttng_strncpy(msg.u.snapshot_channel.pathname,
1530 channel_path,
1531 sizeof(msg.u.snapshot_channel.pathname));
1532 if (ret < 0) {
1533 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1534 sizeof(msg.u.snapshot_channel.pathname),
1535 strlen(channel_path),
1536 channel_path);
1537 status = LTTNG_ERR_SNAPSHOT_FAIL;
1538 goto error;
6dc3064a
DG
1539 }
1540
1541 health_code_update();
9d1103e6 1542 pthread_mutex_lock(socket->lock);
6dc3064a 1543 ret = consumer_send_msg(socket, &msg);
9d1103e6 1544 pthread_mutex_unlock(socket->lock);
6dc3064a 1545 if (ret < 0) {
9bbfb88c
MD
1546 switch (-ret) {
1547 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
9a654598 1548 status = LTTNG_ERR_CHAN_NOT_FOUND;
9bbfb88c
MD
1549 break;
1550 default:
9a654598 1551 status = LTTNG_ERR_SNAPSHOT_FAIL;
9bbfb88c
MD
1552 break;
1553 }
6dc3064a
DG
1554 goto error;
1555 }
1556
1557error:
1558 health_code_update();
9a654598 1559 return status;
6dc3064a 1560}
fb83fe64
JD
1561
1562/*
1563 * Ask the consumer the number of discarded events for a channel.
1564 */
1565int consumer_get_discarded_events(uint64_t session_id, uint64_t channel_key,
1566 struct consumer_output *consumer, uint64_t *discarded)
1567{
1568 int ret;
1569 struct consumer_socket *socket;
1570 struct lttng_ht_iter iter;
1571 struct lttcomm_consumer_msg msg;
1572
1573 assert(consumer);
1574
1575 DBG3("Consumer discarded events id %" PRIu64, session_id);
1576
1577 memset(&msg, 0, sizeof(msg));
1578 msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1579 msg.u.discarded_events.session_id = session_id;
1580 msg.u.discarded_events.channel_key = channel_key;
1581
1582 *discarded = 0;
1583
1584 /* Send command for each consumer */
1585 rcu_read_lock();
1586 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1587 node.node) {
1588 uint64_t consumer_discarded = 0;
1589 pthread_mutex_lock(socket->lock);
1590 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1591 if (ret < 0) {
1592 pthread_mutex_unlock(socket->lock);
1593 goto end;
1594 }
1595
1596 /*
1597 * No need for a recv reply status because the answer to the
1598 * command is the reply status message.
1599 */
1600 ret = consumer_socket_recv(socket, &consumer_discarded,
1601 sizeof(consumer_discarded));
1602 if (ret < 0) {
1603 ERR("get discarded events");
1604 pthread_mutex_unlock(socket->lock);
1605 goto end;
1606 }
1607 pthread_mutex_unlock(socket->lock);
1608 *discarded += consumer_discarded;
1609 }
1610 ret = 0;
1611 DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64,
1612 *discarded, session_id);
1613
1614end:
1615 rcu_read_unlock();
1616 return ret;
1617}
1618
1619/*
1620 * Ask the consumer the number of lost packets for a channel.
1621 */
1622int consumer_get_lost_packets(uint64_t session_id, uint64_t channel_key,
1623 struct consumer_output *consumer, uint64_t *lost)
1624{
1625 int ret;
1626 struct consumer_socket *socket;
1627 struct lttng_ht_iter iter;
1628 struct lttcomm_consumer_msg msg;
1629
1630 assert(consumer);
1631
1632 DBG3("Consumer lost packets id %" PRIu64, session_id);
1633
1634 memset(&msg, 0, sizeof(msg));
1635 msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1636 msg.u.lost_packets.session_id = session_id;
1637 msg.u.lost_packets.channel_key = channel_key;
1638
1639 *lost = 0;
1640
1641 /* Send command for each consumer */
1642 rcu_read_lock();
1643 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1644 node.node) {
1645 uint64_t consumer_lost = 0;
1646 pthread_mutex_lock(socket->lock);
1647 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1648 if (ret < 0) {
1649 pthread_mutex_unlock(socket->lock);
1650 goto end;
1651 }
1652
1653 /*
1654 * No need for a recv reply status because the answer to the
1655 * command is the reply status message.
1656 */
1657 ret = consumer_socket_recv(socket, &consumer_lost,
1658 sizeof(consumer_lost));
1659 if (ret < 0) {
1660 ERR("get lost packets");
1661 pthread_mutex_unlock(socket->lock);
1662 goto end;
1663 }
1664 pthread_mutex_unlock(socket->lock);
1665 *lost += consumer_lost;
1666 }
1667 ret = 0;
1668 DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64,
1669 *lost, session_id);
1670
1671end:
1672 rcu_read_unlock();
1673 return ret;
1674}
a1ae2ea5 1675
5c408ad8
JD
1676/*
1677 * Ask the consumer to rotate a channel.
5c408ad8
JD
1678 *
1679 * The new_chunk_id is the session->rotate_count that has been incremented
1680 * when the rotation started. On the relay, this allows to keep track in which
1681 * chunk each stream is currently writing to (for the rotate_pending operation).
1682 */
1683int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key,
1684 uid_t uid, gid_t gid, struct consumer_output *output,
d2956687 1685 bool is_metadata_channel)
5c408ad8
JD
1686{
1687 int ret;
1688 struct lttcomm_consumer_msg msg;
1689
1690 assert(socket);
1691
1692 DBG("Consumer rotate channel key %" PRIu64, key);
1693
1694 pthread_mutex_lock(socket->lock);
1695 memset(&msg, 0, sizeof(msg));
1696 msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL;
1697 msg.u.rotate_channel.key = key;
1698 msg.u.rotate_channel.metadata = !!is_metadata_channel;
5c408ad8
JD
1699
1700 if (output->type == CONSUMER_DST_NET) {
1701 msg.u.rotate_channel.relayd_id = output->net_seq_index;
5c408ad8
JD
1702 } else {
1703 msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL;
5c408ad8
JD
1704 }
1705
1706 health_code_update();
1707 ret = consumer_send_msg(socket, &msg);
1708 if (ret < 0) {
20f37cb4
MD
1709 switch (-ret) {
1710 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1711 ret = -LTTNG_ERR_CHAN_NOT_FOUND;
1712 break;
1713 default:
1714 ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER;
1715 break;
1716 }
5c408ad8
JD
1717 goto error;
1718 }
5c408ad8
JD
1719error:
1720 pthread_mutex_unlock(socket->lock);
1721 health_code_update();
1722 return ret;
1723}
1724
04ed9e10
JG
1725int consumer_open_channel_packets(struct consumer_socket *socket, uint64_t key)
1726{
1727 int ret;
1728 const struct lttcomm_consumer_msg msg = {
1729 .cmd_type = LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS,
1730 .u.open_channel_packets.key = key,
1731 };
1732
1733 assert(socket);
1734
1735 DBG("Consumer open channel packets: channel key = %" PRIu64, key);
1736
1737 health_code_update();
1738
1739 pthread_mutex_lock(socket->lock);
1740 ret = consumer_send_msg(socket, &msg);
1741 pthread_mutex_unlock(socket->lock);
1742 if (ret < 0) {
1743 goto error_socket;
1744 }
1745
1746error_socket:
1747 health_code_update();
1748 return ret;
1749}
1750
51a4828f
MD
1751int consumer_clear_channel(struct consumer_socket *socket, uint64_t key)
1752{
1753 int ret;
1754 struct lttcomm_consumer_msg msg;
1755
1756 assert(socket);
1757
1758 DBG("Consumer clear channel %" PRIu64, key);
1759
1760 memset(&msg, 0, sizeof(msg));
1761 msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL;
1762 msg.u.clear_channel.key = key;
1763
1764 health_code_update();
1765
1766 pthread_mutex_lock(socket->lock);
1767 ret = consumer_send_msg(socket, &msg);
1768 if (ret < 0) {
1769 goto error_socket;
1770 }
1771
1772error_socket:
1773 pthread_mutex_unlock(socket->lock);
1774
1775 health_code_update();
1776 return ret;
1777}
1778
d2956687
JG
1779int consumer_init(struct consumer_socket *socket,
1780 const lttng_uuid sessiond_uuid)
00fb02ac
JD
1781{
1782 int ret;
d2956687
JG
1783 struct lttcomm_consumer_msg msg = {
1784 .cmd_type = LTTNG_CONSUMER_INIT,
1785 };
00fb02ac
JD
1786
1787 assert(socket);
00fb02ac 1788
d2956687
JG
1789 DBG("Sending consumer initialization command");
1790 lttng_uuid_copy(msg.u.init.sessiond_uuid, sessiond_uuid);
00fb02ac
JD
1791
1792 health_code_update();
1793 ret = consumer_send_msg(socket, &msg);
1794 if (ret < 0) {
1795 goto error;
1796 }
1797
d88744a4
JD
1798error:
1799 health_code_update();
1800 return ret;
1801}
1802
1803/*
d2956687 1804 * Ask the consumer to create a new chunk for a given session.
92816cc3 1805 *
d2956687 1806 * Called with the consumer socket lock held.
92816cc3 1807 */
d2956687
JG
1808int consumer_create_trace_chunk(struct consumer_socket *socket,
1809 uint64_t relayd_id, uint64_t session_id,
5da88b0f
MD
1810 struct lttng_trace_chunk *chunk,
1811 const char *domain_subdir)
92816cc3
JG
1812{
1813 int ret;
d2956687
JG
1814 enum lttng_trace_chunk_status chunk_status;
1815 struct lttng_credentials chunk_credentials;
5da88b0f
MD
1816 const struct lttng_directory_handle *chunk_directory_handle = NULL;
1817 struct lttng_directory_handle *domain_handle = NULL;
1818 int domain_dirfd;
d2956687 1819 const char *chunk_name;
913a542b 1820 bool chunk_name_overridden;
d2956687
JG
1821 uint64_t chunk_id;
1822 time_t creation_timestamp;
1823 char creation_timestamp_buffer[ISO8601_STR_LEN];
1824 const char *creation_timestamp_str = "(none)";
1825 const bool chunk_has_local_output = relayd_id == -1ULL;
69ebf37e 1826 enum lttng_trace_chunk_status tc_status;
d2956687
JG
1827 struct lttcomm_consumer_msg msg = {
1828 .cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
1829 .u.create_trace_chunk.session_id = session_id,
1830 };
92816cc3
JG
1831
1832 assert(socket);
d2956687 1833 assert(chunk);
92816cc3 1834
d2956687
JG
1835 if (relayd_id != -1ULL) {
1836 LTTNG_OPTIONAL_SET(&msg.u.create_trace_chunk.relayd_id,
1837 relayd_id);
1838 }
92816cc3 1839
d2956687 1840 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name,
913a542b 1841 &chunk_name_overridden);
d2956687
JG
1842 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1843 chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1844 ERR("Failed to get name of trace chunk");
1845 ret = -LTTNG_ERR_FATAL;
92816cc3
JG
1846 goto error;
1847 }
913a542b 1848 if (chunk_name_overridden) {
d2956687
JG
1849 ret = lttng_strncpy(msg.u.create_trace_chunk.override_name,
1850 chunk_name,
1851 sizeof(msg.u.create_trace_chunk.override_name));
1852 if (ret) {
1853 ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1854 chunk_name);
1855 ret = -LTTNG_ERR_FATAL;
1856 goto error;
1857 }
1858 }
92816cc3 1859
d2956687
JG
1860 chunk_status = lttng_trace_chunk_get_creation_timestamp(chunk,
1861 &creation_timestamp);
1862 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1863 ret = -LTTNG_ERR_FATAL;
92816cc3
JG
1864 goto error;
1865 }
d2956687
JG
1866 msg.u.create_trace_chunk.creation_timestamp =
1867 (uint64_t) creation_timestamp;
1868 /* Only used for logging purposes. */
1869 ret = time_to_iso8601_str(creation_timestamp,
1870 creation_timestamp_buffer,
1871 sizeof(creation_timestamp_buffer));
1872 creation_timestamp_str = !ret ? creation_timestamp_buffer :
1873 "(formatting error)";
1874
1875 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1876 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1877 /*
1878 * Anonymous trace chunks should never be transmitted
1879 * to remote peers (consumerd and relayd). They are used
1880 * internally for backward-compatibility purposes.
1881 */
1882 ret = -LTTNG_ERR_FATAL;
1883 goto error;
1884 }
1885 msg.u.create_trace_chunk.chunk_id = chunk_id;
92816cc3 1886
d2956687 1887 if (chunk_has_local_output) {
cbf53d23 1888 chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle(
d2956687
JG
1889 chunk, &chunk_directory_handle);
1890 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1891 ret = -LTTNG_ERR_FATAL;
1892 goto error;
1893 }
e5add6d0
JG
1894 chunk_status = lttng_trace_chunk_get_credentials(
1895 chunk, &chunk_credentials);
1896 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1897 /*
1898 * Not associating credentials to a sessiond chunk is a
1899 * fatal internal error.
1900 */
1901 ret = -LTTNG_ERR_FATAL;
1902 goto error;
1903 }
69ebf37e
JR
1904 tc_status = lttng_trace_chunk_create_subdirectory(
1905 chunk, domain_subdir);
1906 if (tc_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
5da88b0f
MD
1907 PERROR("Failed to create chunk domain output directory \"%s\"",
1908 domain_subdir);
1909 ret = -LTTNG_ERR_FATAL;
1910 goto error;
1911 }
1912 domain_handle = lttng_directory_handle_create_from_handle(
1913 domain_subdir,
1914 chunk_directory_handle);
1915 if (!domain_handle) {
1916 ret = -LTTNG_ERR_FATAL;
1917 goto error;
1918 }
1919
1920 /*
1921 * This will only compile on platforms that support
1922 * dirfd (POSIX.2008). This is fine as the session daemon
1923 * is only built for such platforms.
1924 *
1925 * The ownership of the chunk directory handle's is maintained
1926 * by the trace chunk.
1927 */
1928 domain_dirfd = lttng_directory_handle_get_dirfd(
1929 domain_handle);
1930 assert(domain_dirfd >= 0);
1931
e5add6d0
JG
1932 msg.u.create_trace_chunk.credentials.value.uid =
1933 chunk_credentials.uid;
1934 msg.u.create_trace_chunk.credentials.value.gid =
1935 chunk_credentials.gid;
1936 msg.u.create_trace_chunk.credentials.is_set = 1;
d2956687 1937 }
d88744a4 1938
d2956687
JG
1939 DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1940 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
1941 ", creation_timestamp = %s",
1942 relayd_id, session_id, chunk_id,
1943 creation_timestamp_str);
d88744a4
JD
1944 health_code_update();
1945 ret = consumer_send_msg(socket, &msg);
d2956687 1946 health_code_update();
d88744a4 1947 if (ret < 0) {
d2956687
JG
1948 ERR("Trace chunk creation error on consumer");
1949 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
d88744a4
JD
1950 goto error;
1951 }
1952
d2956687 1953 if (chunk_has_local_output) {
5da88b0f 1954 DBG("Sending trace chunk domain directory fd to consumer");
d2956687 1955 health_code_update();
5da88b0f 1956 ret = consumer_send_fds(socket, &domain_dirfd, 1);
d2956687
JG
1957 health_code_update();
1958 if (ret < 0) {
1959 ERR("Trace chunk creation error on consumer");
1960 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1961 goto error;
1962 }
d88744a4 1963 }
00fb02ac 1964error:
5da88b0f 1965 lttng_directory_handle_put(domain_handle);
00fb02ac
JD
1966 return ret;
1967}
1968
a1ae2ea5 1969/*
d2956687 1970 * Ask the consumer to close a trace chunk for a given session.
a1ae2ea5
JD
1971 *
1972 * Called with the consumer socket lock held.
1973 */
d2956687
JG
1974int consumer_close_trace_chunk(struct consumer_socket *socket,
1975 uint64_t relayd_id, uint64_t session_id,
ecd1a12f
MD
1976 struct lttng_trace_chunk *chunk,
1977 char *closed_trace_chunk_path)
a1ae2ea5
JD
1978{
1979 int ret;
d2956687
JG
1980 enum lttng_trace_chunk_status chunk_status;
1981 struct lttcomm_consumer_msg msg = {
bbc4768c
JG
1982 .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
1983 .u.close_trace_chunk.session_id = session_id,
d2956687 1984 };
ecd1a12f 1985 struct lttcomm_consumer_close_trace_chunk_reply reply;
d2956687
JG
1986 uint64_t chunk_id;
1987 time_t close_timestamp;
bbc4768c
JG
1988 enum lttng_trace_chunk_command_type close_command;
1989 const char *close_command_name = "none";
ecd1a12f 1990 struct lttng_dynamic_buffer path_reception_buffer;
a1ae2ea5
JD
1991
1992 assert(socket);
ecd1a12f 1993 lttng_dynamic_buffer_init(&path_reception_buffer);
a1ae2ea5 1994
d2956687 1995 if (relayd_id != -1ULL) {
bbc4768c
JG
1996 LTTNG_OPTIONAL_SET(
1997 &msg.u.close_trace_chunk.relayd_id, relayd_id);
1998 }
1999
2000 chunk_status = lttng_trace_chunk_get_close_command(
2001 chunk, &close_command);
2002 switch (chunk_status) {
2003 case LTTNG_TRACE_CHUNK_STATUS_OK:
2004 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command,
2005 (uint32_t) close_command);
2006 break;
2007 case LTTNG_TRACE_CHUNK_STATUS_NONE:
2008 break;
2009 default:
2010 ERR("Failed to get trace chunk close command");
2011 ret = -1;
2012 goto error;
a1ae2ea5
JD
2013 }
2014
d2956687
JG
2015 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2016 /*
2017 * Anonymous trace chunks should never be transmitted to remote peers
2018 * (consumerd and relayd). They are used internally for
2019 * backward-compatibility purposes.
2020 */
2021 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2022 msg.u.close_trace_chunk.chunk_id = chunk_id;
2023
2024 chunk_status = lttng_trace_chunk_get_close_timestamp(chunk,
2025 &close_timestamp);
2026 /*
2027 * A trace chunk should be closed locally before being closed remotely.
2028 * Otherwise, the close timestamp would never be transmitted to the
2029 * peers.
2030 */
2031 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2032 msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp;
2033
bbc4768c
JG
2034 if (msg.u.close_trace_chunk.close_command.is_set) {
2035 close_command_name = lttng_trace_chunk_command_type_get_name(
2036 close_command);
2037 }
d2956687 2038 DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
bbc4768c
JG
2039 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
2040 ", close command = \"%s\"",
2041 relayd_id, session_id, chunk_id, close_command_name);
a1ae2ea5
JD
2042
2043 health_code_update();
ecd1a12f 2044 ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg));
a1ae2ea5 2045 if (ret < 0) {
d2956687 2046 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
a1ae2ea5
JD
2047 goto error;
2048 }
ecd1a12f
MD
2049 ret = consumer_socket_recv(socket, &reply, sizeof(reply));
2050 if (ret < 0) {
2051 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2052 goto error;
2053 }
2054 if (reply.path_length >= LTTNG_PATH_MAX) {
2055 ERR("Invalid path returned by relay daemon: %" PRIu32 "bytes exceeds maximal allowed length of %d bytes",
2056 reply.path_length, LTTNG_PATH_MAX);
2057 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2058 goto error;
2059 }
2060 ret = lttng_dynamic_buffer_set_size(&path_reception_buffer,
2061 reply.path_length);
2062 if (ret) {
2063 ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command");
2064 ret = -LTTNG_ERR_NOMEM;
2065 goto error;
2066 }
2067 ret = consumer_socket_recv(socket, path_reception_buffer.data,
2068 path_reception_buffer.size);
2069 if (ret < 0) {
2070 ERR("Communication error while receiving path of closed trace chunk");
2071 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2072 goto error;
2073 }
2074 if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') {
2075 ERR("Invalid path returned by relay daemon: not null-terminated");
2076 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2077 goto error;
2078 }
2079 if (closed_trace_chunk_path) {
2080 /*
2081 * closed_trace_chunk_path is assumed to have a length >=
2082 * LTTNG_PATH_MAX
2083 */
2084 memcpy(closed_trace_chunk_path, path_reception_buffer.data,
2085 path_reception_buffer.size);
2086 }
a1ae2ea5 2087error:
ecd1a12f 2088 lttng_dynamic_buffer_reset(&path_reception_buffer);
a1ae2ea5
JD
2089 health_code_update();
2090 return ret;
2091}
3654ed19 2092
d2956687
JG
2093/*
2094 * Ask the consumer if a trace chunk exists.
2095 *
2096 * Called with the consumer socket lock held.
2097 * Returns 0 on success, or a negative value on error.
2098 */
2099int consumer_trace_chunk_exists(struct consumer_socket *socket,
2100 uint64_t relayd_id, uint64_t session_id,
2101 struct lttng_trace_chunk *chunk,
2102 enum consumer_trace_chunk_exists_status *result)
3654ed19
JG
2103{
2104 int ret;
d2956687 2105 enum lttng_trace_chunk_status chunk_status;
3654ed19 2106 struct lttcomm_consumer_msg msg = {
d2956687
JG
2107 .cmd_type = LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,
2108 .u.trace_chunk_exists.session_id = session_id,
3654ed19 2109 };
d2956687
JG
2110 uint64_t chunk_id;
2111 const char *consumer_reply_str;
3654ed19
JG
2112
2113 assert(socket);
2114
d2956687
JG
2115 if (relayd_id != -1ULL) {
2116 LTTNG_OPTIONAL_SET(&msg.u.trace_chunk_exists.relayd_id,
2117 relayd_id);
2118 }
2119
2120 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2121 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2122 /*
2123 * Anonymous trace chunks should never be transmitted
2124 * to remote peers (consumerd and relayd). They are used
2125 * internally for backward-compatibility purposes.
2126 */
2127 ret = -LTTNG_ERR_FATAL;
2128 goto error;
2129 }
2130 msg.u.trace_chunk_exists.chunk_id = chunk_id;
2131
2132 DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
2133 ", session_id = %" PRIu64
2134 ", chunk_id = %" PRIu64, relayd_id, session_id, chunk_id);
3654ed19
JG
2135
2136 health_code_update();
2137 ret = consumer_send_msg(socket, &msg);
d2956687
JG
2138 switch (-ret) {
2139 case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK:
2140 consumer_reply_str = "unknown trace chunk";
2141 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK;
2142 break;
2143 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL:
2144 consumer_reply_str = "trace chunk exists locally";
2145 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL;
2146 break;
2147 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE:
2148 consumer_reply_str = "trace chunk exists on remote peer";
2149 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE;
2150 break;
2151 default:
2152 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
2153 ret = -1;
3654ed19
JG
2154 goto error;
2155 }
d2956687
JG
2156 DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
2157 consumer_reply_str);
2158 ret = 0;
3654ed19
JG
2159error:
2160 health_code_update();
2161 return ret;
2162}
This page took 0.273698 seconds and 4 git commands to generate.