Live timer set up
[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 _GNU_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
32 #include "consumer.h"
33 #include "health.h"
34 #include "ust-app.h"
35 #include "utils.h"
36
37 /*
38 * Send a data payload using a given consumer socket of size len.
39 *
40 * The consumer socket lock MUST be acquired before calling this since this
41 * function can change the fd value.
42 *
43 * Return 0 on success else a negative value on error.
44 */
45 int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len)
46 {
47 int fd;
48 ssize_t size;
49
50 assert(socket);
51 assert(socket->fd_ptr);
52 assert(msg);
53
54 /* Consumer socket is invalid. Stopping. */
55 fd = *socket->fd_ptr;
56 if (fd < 0) {
57 goto error;
58 }
59
60 size = lttcomm_send_unix_sock(fd, msg, len);
61 if (size < 0) {
62 /* The above call will print a PERROR on error. */
63 DBG("Error when sending data to consumer on sock %d", fd);
64 /*
65 * At this point, the socket is not usable anymore thus closing it and
66 * setting the file descriptor to -1 so it is not reused.
67 */
68
69 /* This call will PERROR on error. */
70 (void) lttcomm_close_unix_sock(fd);
71 *socket->fd_ptr = -1;
72 goto error;
73 }
74
75 return 0;
76
77 error:
78 return -1;
79 }
80
81 /*
82 * Receive a data payload using a given consumer socket of size len.
83 *
84 * The consumer socket lock MUST be acquired before calling this since this
85 * function can change the fd value.
86 *
87 * Return 0 on success else a negative value on error.
88 */
89 int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
90 {
91 int fd;
92 ssize_t size;
93
94 assert(socket);
95 assert(socket->fd_ptr);
96 assert(msg);
97
98 /* Consumer socket is invalid. Stopping. */
99 fd = *socket->fd_ptr;
100 if (fd < 0) {
101 goto error;
102 }
103
104 size = lttcomm_recv_unix_sock(fd, msg, len);
105 if (size <= 0) {
106 /* The above call will print a PERROR on error. */
107 DBG("Error when receiving data from the consumer socket %d", fd);
108 /*
109 * At this point, the socket is not usable anymore thus closing it and
110 * setting the file descriptor to -1 so it is not reused.
111 */
112
113 /* This call will PERROR on error. */
114 (void) lttcomm_close_unix_sock(fd);
115 *socket->fd_ptr = -1;
116 goto error;
117 }
118
119 return 0;
120
121 error:
122 return -1;
123 }
124
125 /*
126 * Receive a reply command status message from the consumer. Consumer socket
127 * lock MUST be acquired before calling this function.
128 *
129 * Return 0 on success, -1 on recv error or a negative lttng error code which
130 * was possibly returned by the consumer.
131 */
132 int consumer_recv_status_reply(struct consumer_socket *sock)
133 {
134 int ret;
135 struct lttcomm_consumer_status_msg reply;
136
137 assert(sock);
138
139 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
140 if (ret < 0) {
141 goto end;
142 }
143
144 if (reply.ret_code == LTTNG_OK) {
145 /* All good. */
146 ret = 0;
147 } else {
148 ret = -reply.ret_code;
149 DBG("Consumer ret code %d", ret);
150 }
151
152 end:
153 return ret;
154 }
155
156 /*
157 * Once the ASK_CHANNEL command is sent to the consumer, the channel
158 * information are sent back. This call receives that data and populates key
159 * and stream_count.
160 *
161 * On success return 0 and both key and stream_count are set. On error, a
162 * negative value is sent back and both parameters are untouched.
163 */
164 int consumer_recv_status_channel(struct consumer_socket *sock,
165 uint64_t *key, unsigned int *stream_count)
166 {
167 int ret;
168 struct lttcomm_consumer_status_channel reply;
169
170 assert(sock);
171 assert(stream_count);
172 assert(key);
173
174 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
175 if (ret < 0) {
176 goto end;
177 }
178
179 /* An error is possible so don't touch the key and stream_count. */
180 if (reply.ret_code != LTTNG_OK) {
181 ret = -1;
182 goto end;
183 }
184
185 *key = reply.key;
186 *stream_count = reply.stream_count;
187
188 end:
189 return ret;
190 }
191
192 /*
193 * Send destroy relayd command to consumer.
194 *
195 * On success return positive value. On error, negative value.
196 */
197 int consumer_send_destroy_relayd(struct consumer_socket *sock,
198 struct consumer_output *consumer)
199 {
200 int ret;
201 struct lttcomm_consumer_msg msg;
202
203 assert(consumer);
204 assert(sock);
205
206 DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
207
208 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
209 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
210
211 pthread_mutex_lock(sock->lock);
212 ret = consumer_socket_send(sock, &msg, sizeof(msg));
213 if (ret < 0) {
214 goto error;
215 }
216
217 /* Don't check the return value. The caller will do it. */
218 ret = consumer_recv_status_reply(sock);
219
220 DBG2("Consumer send destroy relayd command done");
221
222 error:
223 pthread_mutex_unlock(sock->lock);
224 return ret;
225 }
226
227 /*
228 * For each consumer socket in the consumer output object, send a destroy
229 * relayd command.
230 */
231 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
232 {
233 struct lttng_ht_iter iter;
234 struct consumer_socket *socket;
235
236 assert(consumer);
237
238 /* Destroy any relayd connection */
239 if (consumer->type == CONSUMER_DST_NET) {
240 rcu_read_lock();
241 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
242 node.node) {
243 int ret;
244
245 /* Send destroy relayd command */
246 ret = consumer_send_destroy_relayd(socket, consumer);
247 if (ret < 0) {
248 DBG("Unable to send destroy relayd command to consumer");
249 /* Continue since we MUST delete everything at this point. */
250 }
251 }
252 rcu_read_unlock();
253 }
254 }
255
256 /*
257 * From a consumer_data structure, allocate and add a consumer socket to the
258 * consumer output.
259 *
260 * Return 0 on success, else negative value on error
261 */
262 int consumer_create_socket(struct consumer_data *data,
263 struct consumer_output *output)
264 {
265 int ret = 0;
266 struct consumer_socket *socket;
267
268 assert(data);
269
270 if (output == NULL || data->cmd_sock < 0) {
271 /*
272 * Not an error. Possible there is simply not spawned consumer or it's
273 * disabled for the tracing session asking the socket.
274 */
275 goto error;
276 }
277
278 rcu_read_lock();
279 socket = consumer_find_socket(data->cmd_sock, output);
280 rcu_read_unlock();
281 if (socket == NULL) {
282 socket = consumer_allocate_socket(&data->cmd_sock);
283 if (socket == NULL) {
284 ret = -1;
285 goto error;
286 }
287
288 socket->registered = 0;
289 socket->lock = &data->lock;
290 rcu_read_lock();
291 consumer_add_socket(socket, output);
292 rcu_read_unlock();
293 }
294
295 socket->type = data->type;
296
297 DBG3("Consumer socket created (fd: %d) and added to output",
298 data->cmd_sock);
299
300 error:
301 return ret;
302 }
303
304 /*
305 * Return the consumer socket from the given consumer output with the right
306 * bitness. On error, returns NULL.
307 *
308 * The caller MUST acquire a rcu read side lock and keep it until the socket
309 * object reference is not needed anymore.
310 */
311 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
312 struct consumer_output *consumer)
313 {
314 int consumer_fd;
315 struct consumer_socket *socket = NULL;
316
317 switch (bits) {
318 case 64:
319 consumer_fd = uatomic_read(&ust_consumerd64_fd);
320 break;
321 case 32:
322 consumer_fd = uatomic_read(&ust_consumerd32_fd);
323 break;
324 default:
325 assert(0);
326 goto end;
327 }
328
329 socket = consumer_find_socket(consumer_fd, consumer);
330 if (!socket) {
331 ERR("Consumer socket fd %d not found in consumer obj %p",
332 consumer_fd, consumer);
333 }
334
335 end:
336 return socket;
337 }
338
339 /*
340 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
341 * be acquired before calling this function and across use of the
342 * returned consumer_socket.
343 */
344 struct consumer_socket *consumer_find_socket(int key,
345 struct consumer_output *consumer)
346 {
347 struct lttng_ht_iter iter;
348 struct lttng_ht_node_ulong *node;
349 struct consumer_socket *socket = NULL;
350
351 /* Negative keys are lookup failures */
352 if (key < 0 || consumer == NULL) {
353 return NULL;
354 }
355
356 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
357 &iter);
358 node = lttng_ht_iter_get_node_ulong(&iter);
359 if (node != NULL) {
360 socket = caa_container_of(node, struct consumer_socket, node);
361 }
362
363 return socket;
364 }
365
366 /*
367 * Allocate a new consumer_socket and return the pointer.
368 */
369 struct consumer_socket *consumer_allocate_socket(int *fd)
370 {
371 struct consumer_socket *socket = NULL;
372
373 assert(fd);
374
375 socket = zmalloc(sizeof(struct consumer_socket));
376 if (socket == NULL) {
377 PERROR("zmalloc consumer socket");
378 goto error;
379 }
380
381 socket->fd_ptr = fd;
382 lttng_ht_node_init_ulong(&socket->node, *fd);
383
384 error:
385 return socket;
386 }
387
388 /*
389 * Add consumer socket to consumer output object. Read side lock must be
390 * acquired before calling this function.
391 */
392 void consumer_add_socket(struct consumer_socket *sock,
393 struct consumer_output *consumer)
394 {
395 assert(sock);
396 assert(consumer);
397
398 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
399 }
400
401 /*
402 * Delte consumer socket to consumer output object. Read side lock must be
403 * acquired before calling this function.
404 */
405 void consumer_del_socket(struct consumer_socket *sock,
406 struct consumer_output *consumer)
407 {
408 int ret;
409 struct lttng_ht_iter iter;
410
411 assert(sock);
412 assert(consumer);
413
414 iter.iter.node = &sock->node.node;
415 ret = lttng_ht_del(consumer->socks, &iter);
416 assert(!ret);
417 }
418
419 /*
420 * RCU destroy call function.
421 */
422 static void destroy_socket_rcu(struct rcu_head *head)
423 {
424 struct lttng_ht_node_ulong *node =
425 caa_container_of(head, struct lttng_ht_node_ulong, head);
426 struct consumer_socket *socket =
427 caa_container_of(node, struct consumer_socket, node);
428
429 free(socket);
430 }
431
432 /*
433 * Destroy and free socket pointer in a call RCU. Read side lock must be
434 * acquired before calling this function.
435 */
436 void consumer_destroy_socket(struct consumer_socket *sock)
437 {
438 assert(sock);
439
440 /*
441 * We DO NOT close the file descriptor here since it is global to the
442 * session daemon and is closed only if the consumer dies or a custom
443 * consumer was registered,
444 */
445 if (sock->registered) {
446 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
447 lttcomm_close_unix_sock(*sock->fd_ptr);
448 }
449
450 call_rcu(&sock->node.head, destroy_socket_rcu);
451 }
452
453 /*
454 * Allocate and assign data to a consumer_output object.
455 *
456 * Return pointer to structure.
457 */
458 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
459 {
460 struct consumer_output *output = NULL;
461
462 output = zmalloc(sizeof(struct consumer_output));
463 if (output == NULL) {
464 PERROR("zmalloc consumer_output");
465 goto error;
466 }
467
468 /* By default, consumer output is enabled */
469 output->enabled = 1;
470 output->type = type;
471 output->net_seq_index = (uint64_t) -1ULL;
472
473 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
474
475 error:
476 return output;
477 }
478
479 /*
480 * Iterate over the consumer output socket hash table and destroy them. The
481 * socket file descriptor are only closed if the consumer output was
482 * registered meaning it's an external consumer.
483 */
484 void consumer_destroy_output_sockets(struct consumer_output *obj)
485 {
486 struct lttng_ht_iter iter;
487 struct consumer_socket *socket;
488
489 if (!obj->socks) {
490 return;
491 }
492
493 rcu_read_lock();
494 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
495 consumer_del_socket(socket, obj);
496 consumer_destroy_socket(socket);
497 }
498 rcu_read_unlock();
499 }
500
501 /*
502 * Delete the consumer_output object from the list and free the ptr.
503 *
504 * Should *NOT* be called with RCU read-side lock held.
505 */
506 void consumer_destroy_output(struct consumer_output *obj)
507 {
508 if (obj == NULL) {
509 return;
510 }
511
512 consumer_destroy_output_sockets(obj);
513
514 if (obj->socks) {
515 /* Finally destroy HT */
516 ht_cleanup_push(obj->socks);
517 }
518
519 free(obj);
520 }
521
522 /*
523 * Copy consumer output and returned the newly allocated copy.
524 *
525 * Should *NOT* be called with RCU read-side lock held.
526 */
527 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
528 {
529 int ret;
530 struct lttng_ht *tmp_ht_ptr;
531 struct consumer_output *output;
532
533 assert(obj);
534
535 output = consumer_create_output(obj->type);
536 if (output == NULL) {
537 goto error;
538 }
539 /* Avoid losing the HT reference after the memcpy() */
540 tmp_ht_ptr = output->socks;
541
542 memcpy(output, obj, sizeof(struct consumer_output));
543
544 /* Putting back the HT pointer and start copying socket(s). */
545 output->socks = tmp_ht_ptr;
546
547 ret = consumer_copy_sockets(output, obj);
548 if (ret < 0) {
549 goto malloc_error;
550 }
551
552 error:
553 return output;
554
555 malloc_error:
556 consumer_destroy_output(output);
557 return NULL;
558 }
559
560 /*
561 * Copy consumer sockets from src to dst.
562 *
563 * Return 0 on success or else a negative value.
564 */
565 int consumer_copy_sockets(struct consumer_output *dst,
566 struct consumer_output *src)
567 {
568 int ret = 0;
569 struct lttng_ht_iter iter;
570 struct consumer_socket *socket, *copy_sock;
571
572 assert(dst);
573 assert(src);
574
575 rcu_read_lock();
576 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
577 /* Ignore socket that are already there. */
578 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
579 if (copy_sock) {
580 continue;
581 }
582
583 /* Create new socket object. */
584 copy_sock = consumer_allocate_socket(socket->fd_ptr);
585 if (copy_sock == NULL) {
586 rcu_read_unlock();
587 ret = -ENOMEM;
588 goto error;
589 }
590
591 copy_sock->registered = socket->registered;
592 /*
593 * This is valid because this lock is shared accross all consumer
594 * object being the global lock of the consumer data structure of the
595 * session daemon.
596 */
597 copy_sock->lock = socket->lock;
598 consumer_add_socket(copy_sock, dst);
599 }
600 rcu_read_unlock();
601
602 error:
603 return ret;
604 }
605
606 /*
607 * Set network URI to the consumer output object.
608 *
609 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
610 * error.
611 */
612 int consumer_set_network_uri(struct consumer_output *obj,
613 struct lttng_uri *uri)
614 {
615 int ret;
616 char tmp_path[PATH_MAX];
617 char hostname[HOST_NAME_MAX];
618 struct lttng_uri *dst_uri = NULL;
619
620 /* Code flow error safety net. */
621 assert(obj);
622 assert(uri);
623
624 switch (uri->stype) {
625 case LTTNG_STREAM_CONTROL:
626 dst_uri = &obj->dst.net.control;
627 obj->dst.net.control_isset = 1;
628 if (uri->port == 0) {
629 /* Assign default port. */
630 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
631 } else {
632 if (obj->dst.net.data_isset && uri->port ==
633 obj->dst.net.data.port) {
634 ret = -LTTNG_ERR_INVALID;
635 goto error;
636 }
637 }
638 DBG3("Consumer control URI set with port %d", uri->port);
639 break;
640 case LTTNG_STREAM_DATA:
641 dst_uri = &obj->dst.net.data;
642 obj->dst.net.data_isset = 1;
643 if (uri->port == 0) {
644 /* Assign default port. */
645 uri->port = DEFAULT_NETWORK_DATA_PORT;
646 } else {
647 if (obj->dst.net.control_isset && uri->port ==
648 obj->dst.net.control.port) {
649 ret = -LTTNG_ERR_INVALID;
650 goto error;
651 }
652 }
653 DBG3("Consumer data URI set with port %d", uri->port);
654 break;
655 default:
656 ERR("Set network uri type unknown %d", uri->stype);
657 ret = -LTTNG_ERR_INVALID;
658 goto error;
659 }
660
661 ret = uri_compare(dst_uri, uri);
662 if (!ret) {
663 /* Same URI, don't touch it and return success. */
664 DBG3("URI network compare are the same");
665 goto equal;
666 }
667
668 /* URIs were not equal, replacing it. */
669 memset(dst_uri, 0, sizeof(struct lttng_uri));
670 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
671 obj->type = CONSUMER_DST_NET;
672
673 /* Handle subdir and add hostname in front. */
674 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
675 /* Get hostname to append it in the pathname */
676 ret = gethostname(hostname, sizeof(hostname));
677 if (ret < 0) {
678 PERROR("gethostname. Fallback on default localhost");
679 strncpy(hostname, "localhost", sizeof(hostname));
680 }
681 hostname[sizeof(hostname) - 1] = '\0';
682
683 /* Setup consumer subdir if none present in the control URI */
684 if (strlen(dst_uri->subdir) == 0) {
685 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
686 hostname, obj->subdir);
687 } else {
688 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
689 hostname, dst_uri->subdir);
690 }
691 if (ret < 0) {
692 PERROR("snprintf set consumer uri subdir");
693 ret = -LTTNG_ERR_NOMEM;
694 goto error;
695 }
696
697 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
698 DBG3("Consumer set network uri subdir path %s", tmp_path);
699 }
700
701 return 0;
702 equal:
703 return 1;
704 error:
705 return ret;
706 }
707
708 /*
709 * Send file descriptor to consumer via sock.
710 */
711 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
712 {
713 int ret;
714
715 assert(fds);
716 assert(sock);
717 assert(nb_fd > 0);
718
719 ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
720 if (ret < 0) {
721 /* The above call will print a PERROR on error. */
722 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
723 goto error;
724 }
725
726 ret = consumer_recv_status_reply(sock);
727
728 error:
729 return ret;
730 }
731
732 /*
733 * Consumer send communication message structure to consumer.
734 */
735 int consumer_send_msg(struct consumer_socket *sock,
736 struct lttcomm_consumer_msg *msg)
737 {
738 int ret;
739
740 assert(msg);
741 assert(sock);
742
743 ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
744 if (ret < 0) {
745 goto error;
746 }
747
748 ret = consumer_recv_status_reply(sock);
749
750 error:
751 return ret;
752 }
753
754 /*
755 * Consumer send channel communication message structure to consumer.
756 */
757 int consumer_send_channel(struct consumer_socket *sock,
758 struct lttcomm_consumer_msg *msg)
759 {
760 int ret;
761
762 assert(msg);
763 assert(sock);
764
765 ret = consumer_send_msg(sock, msg);
766 if (ret < 0) {
767 goto error;
768 }
769
770 error:
771 return ret;
772 }
773
774 /*
775 * Populate the given consumer msg structure with the ask_channel command
776 * information.
777 */
778 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
779 uint64_t subbuf_size,
780 uint64_t num_subbuf,
781 int overwrite,
782 unsigned int switch_timer_interval,
783 unsigned int read_timer_interval,
784 unsigned int live_timer_interval,
785 int output,
786 int type,
787 uint64_t session_id,
788 const char *pathname,
789 const char *name,
790 uid_t uid,
791 gid_t gid,
792 uint64_t relayd_id,
793 uint64_t key,
794 unsigned char *uuid,
795 uint32_t chan_id,
796 uint64_t tracefile_size,
797 uint64_t tracefile_count,
798 uint64_t session_id_per_pid,
799 unsigned int monitor,
800 uint32_t ust_app_uid)
801 {
802 assert(msg);
803
804 /* Zeroed structure */
805 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
806
807 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
808 msg->u.ask_channel.subbuf_size = subbuf_size;
809 msg->u.ask_channel.num_subbuf = num_subbuf ;
810 msg->u.ask_channel.overwrite = overwrite;
811 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
812 msg->u.ask_channel.read_timer_interval = read_timer_interval;
813 msg->u.ask_channel.live_timer_interval = live_timer_interval;
814 msg->u.ask_channel.output = output;
815 msg->u.ask_channel.type = type;
816 msg->u.ask_channel.session_id = session_id;
817 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
818 msg->u.ask_channel.uid = uid;
819 msg->u.ask_channel.gid = gid;
820 msg->u.ask_channel.relayd_id = relayd_id;
821 msg->u.ask_channel.key = key;
822 msg->u.ask_channel.chan_id = chan_id;
823 msg->u.ask_channel.tracefile_size = tracefile_size;
824 msg->u.ask_channel.tracefile_count = tracefile_count;
825 msg->u.ask_channel.monitor = monitor;
826 msg->u.ask_channel.ust_app_uid = ust_app_uid;
827
828 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
829
830 if (pathname) {
831 strncpy(msg->u.ask_channel.pathname, pathname,
832 sizeof(msg->u.ask_channel.pathname));
833 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
834 }
835
836 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
837 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
838 }
839
840 /*
841 * Init channel communication message structure.
842 */
843 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
844 enum lttng_consumer_command cmd,
845 uint64_t channel_key,
846 uint64_t session_id,
847 const char *pathname,
848 uid_t uid,
849 gid_t gid,
850 uint64_t relayd_id,
851 const char *name,
852 unsigned int nb_init_streams,
853 enum lttng_event_output output,
854 int type,
855 uint64_t tracefile_size,
856 uint64_t tracefile_count,
857 unsigned int monitor,
858 unsigned int live_timer_interval)
859 {
860 assert(msg);
861
862 /* Zeroed structure */
863 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
864
865 /* Send channel */
866 msg->cmd_type = cmd;
867 msg->u.channel.channel_key = channel_key;
868 msg->u.channel.session_id = session_id;
869 msg->u.channel.uid = uid;
870 msg->u.channel.gid = gid;
871 msg->u.channel.relayd_id = relayd_id;
872 msg->u.channel.nb_init_streams = nb_init_streams;
873 msg->u.channel.output = output;
874 msg->u.channel.type = type;
875 msg->u.channel.tracefile_size = tracefile_size;
876 msg->u.channel.tracefile_count = tracefile_count;
877 msg->u.channel.monitor = monitor;
878 msg->u.channel.live_timer_interval = live_timer_interval;
879
880 strncpy(msg->u.channel.pathname, pathname,
881 sizeof(msg->u.channel.pathname));
882 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
883
884 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
885 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
886 }
887
888 /*
889 * Init stream communication message structure.
890 */
891 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
892 enum lttng_consumer_command cmd,
893 uint64_t channel_key,
894 uint64_t stream_key,
895 int cpu)
896 {
897 assert(msg);
898
899 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
900
901 msg->cmd_type = cmd;
902 msg->u.stream.channel_key = channel_key;
903 msg->u.stream.stream_key = stream_key;
904 msg->u.stream.cpu = cpu;
905 }
906
907 /*
908 * Send stream communication structure to the consumer.
909 */
910 int consumer_send_stream(struct consumer_socket *sock,
911 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
912 int *fds, size_t nb_fd)
913 {
914 int ret;
915
916 assert(msg);
917 assert(dst);
918 assert(sock);
919 assert(fds);
920
921 ret = consumer_send_msg(sock, msg);
922 if (ret < 0) {
923 goto error;
924 }
925
926 ret = consumer_send_fds(sock, fds, nb_fd);
927 if (ret < 0) {
928 goto error;
929 }
930
931 error:
932 return ret;
933 }
934
935 /*
936 * Send relayd socket to consumer associated with a session name.
937 *
938 * On success return positive value. On error, negative value.
939 */
940 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
941 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
942 enum lttng_stream_type type, uint64_t session_id)
943 {
944 int ret;
945 struct lttcomm_consumer_msg msg;
946
947 /* Code flow error. Safety net. */
948 assert(rsock);
949 assert(consumer);
950 assert(consumer_sock);
951
952 /* Bail out if consumer is disabled */
953 if (!consumer->enabled) {
954 ret = LTTNG_OK;
955 goto error;
956 }
957
958 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
959 /*
960 * Assign network consumer output index using the temporary consumer since
961 * this call should only be made from within a set_consumer_uri() function
962 * call in the session daemon.
963 */
964 msg.u.relayd_sock.net_index = consumer->net_seq_index;
965 msg.u.relayd_sock.type = type;
966 msg.u.relayd_sock.session_id = session_id;
967 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
968
969 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
970 ret = consumer_send_msg(consumer_sock, &msg);
971 if (ret < 0) {
972 goto error;
973 }
974
975 DBG3("Sending relayd socket file descriptor to consumer");
976 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
977 if (ret < 0) {
978 goto error;
979 }
980
981 DBG2("Consumer relayd socket sent");
982
983 error:
984 return ret;
985 }
986
987 /*
988 * Set consumer subdirectory using the session name and a generated datetime if
989 * needed. This is appended to the current subdirectory.
990 */
991 int consumer_set_subdir(struct consumer_output *consumer,
992 const char *session_name)
993 {
994 int ret = 0;
995 unsigned int have_default_name = 0;
996 char datetime[16], tmp_path[PATH_MAX];
997 time_t rawtime;
998 struct tm *timeinfo;
999
1000 assert(consumer);
1001 assert(session_name);
1002
1003 memset(tmp_path, 0, sizeof(tmp_path));
1004
1005 /* Flag if we have a default session. */
1006 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
1007 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
1008 have_default_name = 1;
1009 } else {
1010 /* Get date and time for session path */
1011 time(&rawtime);
1012 timeinfo = localtime(&rawtime);
1013 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
1014 }
1015
1016 if (have_default_name) {
1017 ret = snprintf(tmp_path, sizeof(tmp_path),
1018 "%s/%s", consumer->subdir, session_name);
1019 } else {
1020 ret = snprintf(tmp_path, sizeof(tmp_path),
1021 "%s/%s-%s/", consumer->subdir, session_name, datetime);
1022 }
1023 if (ret < 0) {
1024 PERROR("snprintf session name date");
1025 goto error;
1026 }
1027
1028 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
1029 DBG2("Consumer subdir set to %s", consumer->subdir);
1030
1031 error:
1032 return ret;
1033 }
1034
1035 /*
1036 * Ask the consumer if the data is ready to read (NOT pending) for the specific
1037 * session id.
1038 *
1039 * This function has a different behavior with the consumer i.e. that it waits
1040 * for a reply from the consumer if yes or no the data is pending.
1041 */
1042 int consumer_is_data_pending(uint64_t session_id,
1043 struct consumer_output *consumer)
1044 {
1045 int ret;
1046 int32_t ret_code = 0; /* Default is that the data is NOT pending */
1047 struct consumer_socket *socket;
1048 struct lttng_ht_iter iter;
1049 struct lttcomm_consumer_msg msg;
1050
1051 assert(consumer);
1052
1053 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1054
1055 msg.u.data_pending.session_id = session_id;
1056
1057 DBG3("Consumer data pending for id %" PRIu64, session_id);
1058
1059 /* Send command for each consumer */
1060 rcu_read_lock();
1061 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1062 node.node) {
1063 pthread_mutex_lock(socket->lock);
1064 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1065 if (ret < 0) {
1066 pthread_mutex_unlock(socket->lock);
1067 goto error_unlock;
1068 }
1069
1070 /*
1071 * No need for a recv reply status because the answer to the command is
1072 * the reply status message.
1073 */
1074
1075 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1076 if (ret < 0) {
1077 pthread_mutex_unlock(socket->lock);
1078 goto error_unlock;
1079 }
1080 pthread_mutex_unlock(socket->lock);
1081
1082 if (ret_code == 1) {
1083 break;
1084 }
1085 }
1086 rcu_read_unlock();
1087
1088 DBG("Consumer data is %s pending for session id %" PRIu64,
1089 ret_code == 1 ? "" : "NOT", session_id);
1090 return ret_code;
1091
1092 error_unlock:
1093 rcu_read_unlock();
1094 return -1;
1095 }
1096
1097 /*
1098 * Send a flush command to consumer using the given channel key.
1099 *
1100 * Return 0 on success else a negative value.
1101 */
1102 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1103 {
1104 int ret;
1105 struct lttcomm_consumer_msg msg;
1106
1107 assert(socket);
1108
1109 DBG2("Consumer flush channel key %" PRIu64, key);
1110
1111 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1112 msg.u.flush_channel.key = key;
1113
1114 pthread_mutex_lock(socket->lock);
1115 health_code_update();
1116
1117 ret = consumer_send_msg(socket, &msg);
1118 if (ret < 0) {
1119 goto end;
1120 }
1121
1122 end:
1123 health_code_update();
1124 pthread_mutex_unlock(socket->lock);
1125 return ret;
1126 }
1127
1128 /*
1129 * Send a close metdata command to consumer using the given channel key.
1130 *
1131 * Return 0 on success else a negative value.
1132 */
1133 int consumer_close_metadata(struct consumer_socket *socket,
1134 uint64_t metadata_key)
1135 {
1136 int ret;
1137 struct lttcomm_consumer_msg msg;
1138
1139 assert(socket);
1140
1141 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1142
1143 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1144 msg.u.close_metadata.key = metadata_key;
1145
1146 pthread_mutex_lock(socket->lock);
1147 health_code_update();
1148
1149 ret = consumer_send_msg(socket, &msg);
1150 if (ret < 0) {
1151 goto end;
1152 }
1153
1154 end:
1155 health_code_update();
1156 pthread_mutex_unlock(socket->lock);
1157 return ret;
1158 }
1159
1160 /*
1161 * Send a setup metdata command to consumer using the given channel key.
1162 *
1163 * Return 0 on success else a negative value.
1164 */
1165 int consumer_setup_metadata(struct consumer_socket *socket,
1166 uint64_t metadata_key)
1167 {
1168 int ret;
1169 struct lttcomm_consumer_msg msg;
1170
1171 assert(socket);
1172
1173 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1174
1175 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1176 msg.u.setup_metadata.key = metadata_key;
1177
1178 pthread_mutex_lock(socket->lock);
1179 health_code_update();
1180
1181 ret = consumer_send_msg(socket, &msg);
1182 if (ret < 0) {
1183 goto end;
1184 }
1185
1186 end:
1187 health_code_update();
1188 pthread_mutex_unlock(socket->lock);
1189 return ret;
1190 }
1191
1192 /*
1193 * Send metadata string to consumer. Socket lock MUST be acquired.
1194 *
1195 * Return 0 on success else a negative value.
1196 */
1197 int consumer_push_metadata(struct consumer_socket *socket,
1198 uint64_t metadata_key, char *metadata_str, size_t len,
1199 size_t target_offset)
1200 {
1201 int ret;
1202 struct lttcomm_consumer_msg msg;
1203
1204 assert(socket);
1205
1206 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1207
1208 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1209 msg.u.push_metadata.key = metadata_key;
1210 msg.u.push_metadata.target_offset = target_offset;
1211 msg.u.push_metadata.len = len;
1212
1213 health_code_update();
1214 ret = consumer_send_msg(socket, &msg);
1215 if (ret < 0 || len == 0) {
1216 goto end;
1217 }
1218
1219 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1220 len);
1221
1222 ret = consumer_socket_send(socket, metadata_str, len);
1223 if (ret < 0) {
1224 goto end;
1225 }
1226
1227 health_code_update();
1228 ret = consumer_recv_status_reply(socket);
1229 if (ret < 0) {
1230 goto end;
1231 }
1232
1233 end:
1234 health_code_update();
1235 return ret;
1236 }
1237
1238 /*
1239 * Ask the consumer to snapshot a specific channel using the key.
1240 *
1241 * Return 0 on success or else a negative error.
1242 */
1243 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1244 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1245 const char *session_path, int wait, int max_stream_size)
1246 {
1247 int ret;
1248 struct lttcomm_consumer_msg msg;
1249
1250 assert(socket);
1251 assert(output);
1252 assert(output->consumer);
1253
1254 DBG("Consumer snapshot channel key %" PRIu64, key);
1255
1256 memset(&msg, 0, sizeof(msg));
1257 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1258 msg.u.snapshot_channel.key = key;
1259 msg.u.snapshot_channel.max_stream_size = max_stream_size;
1260 msg.u.snapshot_channel.metadata = metadata;
1261
1262 if (output->consumer->type == CONSUMER_DST_NET) {
1263 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1264 msg.u.snapshot_channel.use_relayd = 1;
1265 ret = snprintf(msg.u.snapshot_channel.pathname,
1266 sizeof(msg.u.snapshot_channel.pathname),
1267 "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir,
1268 output->name, output->datetime, output->nb_snapshot,
1269 session_path);
1270 if (ret < 0) {
1271 ret = -LTTNG_ERR_NOMEM;
1272 goto error;
1273 }
1274 } else {
1275 ret = snprintf(msg.u.snapshot_channel.pathname,
1276 sizeof(msg.u.snapshot_channel.pathname),
1277 "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path,
1278 output->name, output->datetime, output->nb_snapshot,
1279 session_path);
1280 if (ret < 0) {
1281 ret = -LTTNG_ERR_NOMEM;
1282 goto error;
1283 }
1284 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1285
1286 /* Create directory. Ignore if exist. */
1287 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1288 S_IRWXU | S_IRWXG, uid, gid);
1289 if (ret < 0) {
1290 if (ret != -EEXIST) {
1291 ERR("Trace directory creation error");
1292 goto error;
1293 }
1294 }
1295 }
1296
1297 health_code_update();
1298 ret = consumer_send_msg(socket, &msg);
1299 if (ret < 0) {
1300 goto error;
1301 }
1302
1303 error:
1304 health_code_update();
1305 return ret;
1306 }
This page took 0.093902 seconds and 4 git commands to generate.