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