8f05476a7fe606935d8f3fcca23d75b2e945bb43
[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 int output,
785 int type,
786 uint64_t session_id,
787 const char *pathname,
788 const char *name,
789 uid_t uid,
790 gid_t gid,
791 uint64_t relayd_id,
792 uint64_t key,
793 unsigned char *uuid,
794 uint32_t chan_id,
795 uint64_t tracefile_size,
796 uint64_t tracefile_count,
797 uint64_t session_id_per_pid,
798 unsigned int monitor,
799 uint32_t ust_app_uid)
800 {
801 assert(msg);
802
803 /* Zeroed structure */
804 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
805
806 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
807 msg->u.ask_channel.subbuf_size = subbuf_size;
808 msg->u.ask_channel.num_subbuf = num_subbuf ;
809 msg->u.ask_channel.overwrite = overwrite;
810 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
811 msg->u.ask_channel.read_timer_interval = read_timer_interval;
812 msg->u.ask_channel.output = output;
813 msg->u.ask_channel.type = type;
814 msg->u.ask_channel.session_id = session_id;
815 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
816 msg->u.ask_channel.uid = uid;
817 msg->u.ask_channel.gid = gid;
818 msg->u.ask_channel.relayd_id = relayd_id;
819 msg->u.ask_channel.key = key;
820 msg->u.ask_channel.chan_id = chan_id;
821 msg->u.ask_channel.tracefile_size = tracefile_size;
822 msg->u.ask_channel.tracefile_count = tracefile_count;
823 msg->u.ask_channel.monitor = monitor;
824 msg->u.ask_channel.ust_app_uid = ust_app_uid;
825
826 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
827
828 if (pathname) {
829 strncpy(msg->u.ask_channel.pathname, pathname,
830 sizeof(msg->u.ask_channel.pathname));
831 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
832 }
833
834 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
835 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
836 }
837
838 /*
839 * Init channel communication message structure.
840 */
841 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
842 enum lttng_consumer_command cmd,
843 uint64_t channel_key,
844 uint64_t session_id,
845 const char *pathname,
846 uid_t uid,
847 gid_t gid,
848 uint64_t relayd_id,
849 const char *name,
850 unsigned int nb_init_streams,
851 enum lttng_event_output output,
852 int type,
853 uint64_t tracefile_size,
854 uint64_t tracefile_count,
855 unsigned int monitor)
856 {
857 assert(msg);
858
859 /* Zeroed structure */
860 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
861
862 /* Send channel */
863 msg->cmd_type = cmd;
864 msg->u.channel.channel_key = channel_key;
865 msg->u.channel.session_id = session_id;
866 msg->u.channel.uid = uid;
867 msg->u.channel.gid = gid;
868 msg->u.channel.relayd_id = relayd_id;
869 msg->u.channel.nb_init_streams = nb_init_streams;
870 msg->u.channel.output = output;
871 msg->u.channel.type = type;
872 msg->u.channel.tracefile_size = tracefile_size;
873 msg->u.channel.tracefile_count = tracefile_count;
874 msg->u.channel.monitor = monitor;
875
876 strncpy(msg->u.channel.pathname, pathname,
877 sizeof(msg->u.channel.pathname));
878 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
879
880 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
881 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
882 }
883
884 /*
885 * Init stream communication message structure.
886 */
887 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
888 enum lttng_consumer_command cmd,
889 uint64_t channel_key,
890 uint64_t stream_key,
891 int cpu)
892 {
893 assert(msg);
894
895 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
896
897 msg->cmd_type = cmd;
898 msg->u.stream.channel_key = channel_key;
899 msg->u.stream.stream_key = stream_key;
900 msg->u.stream.cpu = cpu;
901 }
902
903 /*
904 * Send stream communication structure to the consumer.
905 */
906 int consumer_send_stream(struct consumer_socket *sock,
907 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
908 int *fds, size_t nb_fd)
909 {
910 int ret;
911
912 assert(msg);
913 assert(dst);
914 assert(sock);
915 assert(fds);
916
917 ret = consumer_send_msg(sock, msg);
918 if (ret < 0) {
919 goto error;
920 }
921
922 ret = consumer_send_fds(sock, fds, nb_fd);
923 if (ret < 0) {
924 goto error;
925 }
926
927 error:
928 return ret;
929 }
930
931 /*
932 * Send relayd socket to consumer associated with a session name.
933 *
934 * On success return positive value. On error, negative value.
935 */
936 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
937 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
938 enum lttng_stream_type type, uint64_t session_id)
939 {
940 int ret;
941 struct lttcomm_consumer_msg msg;
942
943 /* Code flow error. Safety net. */
944 assert(rsock);
945 assert(consumer);
946 assert(consumer_sock);
947
948 /* Bail out if consumer is disabled */
949 if (!consumer->enabled) {
950 ret = LTTNG_OK;
951 goto error;
952 }
953
954 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
955 /*
956 * Assign network consumer output index using the temporary consumer since
957 * this call should only be made from within a set_consumer_uri() function
958 * call in the session daemon.
959 */
960 msg.u.relayd_sock.net_index = consumer->net_seq_index;
961 msg.u.relayd_sock.type = type;
962 msg.u.relayd_sock.session_id = session_id;
963 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
964
965 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
966 ret = consumer_send_msg(consumer_sock, &msg);
967 if (ret < 0) {
968 goto error;
969 }
970
971 DBG3("Sending relayd socket file descriptor to consumer");
972 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
973 if (ret < 0) {
974 goto error;
975 }
976
977 DBG2("Consumer relayd socket sent");
978
979 error:
980 return ret;
981 }
982
983 /*
984 * Set consumer subdirectory using the session name and a generated datetime if
985 * needed. This is appended to the current subdirectory.
986 */
987 int consumer_set_subdir(struct consumer_output *consumer,
988 const char *session_name)
989 {
990 int ret = 0;
991 unsigned int have_default_name = 0;
992 char datetime[16], tmp_path[PATH_MAX];
993 time_t rawtime;
994 struct tm *timeinfo;
995
996 assert(consumer);
997 assert(session_name);
998
999 memset(tmp_path, 0, sizeof(tmp_path));
1000
1001 /* Flag if we have a default session. */
1002 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
1003 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
1004 have_default_name = 1;
1005 } else {
1006 /* Get date and time for session path */
1007 time(&rawtime);
1008 timeinfo = localtime(&rawtime);
1009 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
1010 }
1011
1012 if (have_default_name) {
1013 ret = snprintf(tmp_path, sizeof(tmp_path),
1014 "%s/%s", consumer->subdir, session_name);
1015 } else {
1016 ret = snprintf(tmp_path, sizeof(tmp_path),
1017 "%s/%s-%s/", consumer->subdir, session_name, datetime);
1018 }
1019 if (ret < 0) {
1020 PERROR("snprintf session name date");
1021 goto error;
1022 }
1023
1024 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
1025 DBG2("Consumer subdir set to %s", consumer->subdir);
1026
1027 error:
1028 return ret;
1029 }
1030
1031 /*
1032 * Ask the consumer if the data is ready to read (NOT pending) for the specific
1033 * session id.
1034 *
1035 * This function has a different behavior with the consumer i.e. that it waits
1036 * for a reply from the consumer if yes or no the data is pending.
1037 */
1038 int consumer_is_data_pending(uint64_t session_id,
1039 struct consumer_output *consumer)
1040 {
1041 int ret;
1042 int32_t ret_code = 0; /* Default is that the data is NOT pending */
1043 struct consumer_socket *socket;
1044 struct lttng_ht_iter iter;
1045 struct lttcomm_consumer_msg msg;
1046
1047 assert(consumer);
1048
1049 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1050
1051 msg.u.data_pending.session_id = session_id;
1052
1053 DBG3("Consumer data pending for id %" PRIu64, session_id);
1054
1055 /* Send command for each consumer */
1056 rcu_read_lock();
1057 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1058 node.node) {
1059 pthread_mutex_lock(socket->lock);
1060 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1061 if (ret < 0) {
1062 pthread_mutex_unlock(socket->lock);
1063 goto error_unlock;
1064 }
1065
1066 /*
1067 * No need for a recv reply status because the answer to the command is
1068 * the reply status message.
1069 */
1070
1071 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1072 if (ret < 0) {
1073 pthread_mutex_unlock(socket->lock);
1074 goto error_unlock;
1075 }
1076 pthread_mutex_unlock(socket->lock);
1077
1078 if (ret_code == 1) {
1079 break;
1080 }
1081 }
1082 rcu_read_unlock();
1083
1084 DBG("Consumer data is %s pending for session id %" PRIu64,
1085 ret_code == 1 ? "" : "NOT", session_id);
1086 return ret_code;
1087
1088 error_unlock:
1089 rcu_read_unlock();
1090 return -1;
1091 }
1092
1093 /*
1094 * Send a flush command to consumer using the given channel key.
1095 *
1096 * Return 0 on success else a negative value.
1097 */
1098 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1099 {
1100 int ret;
1101 struct lttcomm_consumer_msg msg;
1102
1103 assert(socket);
1104
1105 DBG2("Consumer flush channel key %" PRIu64, key);
1106
1107 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1108 msg.u.flush_channel.key = key;
1109
1110 pthread_mutex_lock(socket->lock);
1111 health_code_update();
1112
1113 ret = consumer_send_msg(socket, &msg);
1114 if (ret < 0) {
1115 goto end;
1116 }
1117
1118 end:
1119 health_code_update();
1120 pthread_mutex_unlock(socket->lock);
1121 return ret;
1122 }
1123
1124 /*
1125 * Send a close metdata command to consumer using the given channel key.
1126 *
1127 * Return 0 on success else a negative value.
1128 */
1129 int consumer_close_metadata(struct consumer_socket *socket,
1130 uint64_t metadata_key)
1131 {
1132 int ret;
1133 struct lttcomm_consumer_msg msg;
1134
1135 assert(socket);
1136
1137 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1138
1139 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1140 msg.u.close_metadata.key = metadata_key;
1141
1142 pthread_mutex_lock(socket->lock);
1143 health_code_update();
1144
1145 ret = consumer_send_msg(socket, &msg);
1146 if (ret < 0) {
1147 goto end;
1148 }
1149
1150 end:
1151 health_code_update();
1152 pthread_mutex_unlock(socket->lock);
1153 return ret;
1154 }
1155
1156 /*
1157 * Send a setup metdata command to consumer using the given channel key.
1158 *
1159 * Return 0 on success else a negative value.
1160 */
1161 int consumer_setup_metadata(struct consumer_socket *socket,
1162 uint64_t metadata_key)
1163 {
1164 int ret;
1165 struct lttcomm_consumer_msg msg;
1166
1167 assert(socket);
1168
1169 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1170
1171 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1172 msg.u.setup_metadata.key = metadata_key;
1173
1174 pthread_mutex_lock(socket->lock);
1175 health_code_update();
1176
1177 ret = consumer_send_msg(socket, &msg);
1178 if (ret < 0) {
1179 goto end;
1180 }
1181
1182 end:
1183 health_code_update();
1184 pthread_mutex_unlock(socket->lock);
1185 return ret;
1186 }
1187
1188 /*
1189 * Send metadata string to consumer. Socket lock MUST be acquired.
1190 *
1191 * Return 0 on success else a negative value.
1192 */
1193 int consumer_push_metadata(struct consumer_socket *socket,
1194 uint64_t metadata_key, char *metadata_str, size_t len,
1195 size_t target_offset)
1196 {
1197 int ret;
1198 struct lttcomm_consumer_msg msg;
1199
1200 assert(socket);
1201
1202 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1203
1204 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1205 msg.u.push_metadata.key = metadata_key;
1206 msg.u.push_metadata.target_offset = target_offset;
1207 msg.u.push_metadata.len = len;
1208
1209 health_code_update();
1210 ret = consumer_send_msg(socket, &msg);
1211 if (ret < 0 || len == 0) {
1212 goto end;
1213 }
1214
1215 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1216 len);
1217
1218 ret = consumer_socket_send(socket, metadata_str, len);
1219 if (ret < 0) {
1220 goto end;
1221 }
1222
1223 health_code_update();
1224 ret = consumer_recv_status_reply(socket);
1225 if (ret < 0) {
1226 goto end;
1227 }
1228
1229 end:
1230 health_code_update();
1231 return ret;
1232 }
1233
1234 /*
1235 * Ask the consumer to snapshot a specific channel using the key.
1236 *
1237 * Return 0 on success or else a negative error.
1238 */
1239 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1240 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1241 const char *session_path, int wait, int max_stream_size)
1242 {
1243 int ret;
1244 struct lttcomm_consumer_msg msg;
1245
1246 assert(socket);
1247 assert(output);
1248 assert(output->consumer);
1249
1250 DBG("Consumer snapshot channel key %" PRIu64, key);
1251
1252 memset(&msg, 0, sizeof(msg));
1253 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1254 msg.u.snapshot_channel.key = key;
1255 msg.u.snapshot_channel.max_stream_size = max_stream_size;
1256 msg.u.snapshot_channel.metadata = metadata;
1257
1258 if (output->consumer->type == CONSUMER_DST_NET) {
1259 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1260 msg.u.snapshot_channel.use_relayd = 1;
1261 ret = snprintf(msg.u.snapshot_channel.pathname,
1262 sizeof(msg.u.snapshot_channel.pathname),
1263 "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir,
1264 output->name, output->datetime, output->nb_snapshot,
1265 session_path);
1266 if (ret < 0) {
1267 ret = -LTTNG_ERR_NOMEM;
1268 goto error;
1269 }
1270 } else {
1271 ret = snprintf(msg.u.snapshot_channel.pathname,
1272 sizeof(msg.u.snapshot_channel.pathname),
1273 "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path,
1274 output->name, output->datetime, output->nb_snapshot,
1275 session_path);
1276 if (ret < 0) {
1277 ret = -LTTNG_ERR_NOMEM;
1278 goto error;
1279 }
1280 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1281
1282 /* Create directory. Ignore if exist. */
1283 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1284 S_IRWXU | S_IRWXG, uid, gid);
1285 if (ret < 0) {
1286 if (ret != -EEXIST) {
1287 ERR("Trace directory creation error");
1288 goto error;
1289 }
1290 }
1291 }
1292
1293 health_code_update();
1294 ret = consumer_send_msg(socket, &msg);
1295 if (ret < 0) {
1296 goto error;
1297 }
1298
1299 error:
1300 health_code_update();
1301 return ret;
1302 }
This page took 0.091452 seconds and 3 git commands to generate.