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