Consumer daemon data available command support
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
CommitLineData
00e2e675
DG
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
27#include <common/common.h>
28#include <common/defaults.h>
29#include <common/uri.h>
30
31#include "consumer.h"
32
2f77fc4b
DG
33/*
34 * Send destroy relayd command to consumer.
35 *
36 * On success return positive value. On error, negative value.
37 */
38int consumer_send_destroy_relayd(struct consumer_socket *sock,
39 struct consumer_output *consumer)
40{
41 int ret;
42 struct lttcomm_consumer_msg msg;
43
44 assert(consumer);
45 assert(sock);
46
47 DBG2("Sending destroy relayd command to consumer...");
48
49 /* Bail out if consumer is disabled */
50 if (!consumer->enabled) {
f73fabfd 51 ret = LTTNG_OK;
2f77fc4b
DG
52 DBG3("Consumer is disabled");
53 goto error;
54 }
55
56 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
57 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
58
59 pthread_mutex_lock(sock->lock);
60 ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
61 pthread_mutex_unlock(sock->lock);
62 if (ret < 0) {
63 PERROR("send consumer destroy relayd command");
64 goto error;
65 }
66
67 DBG2("Consumer send destroy relayd command done");
68
69error:
70 return ret;
71}
72
73/*
74 * For each consumer socket in the consumer output object, send a destroy
75 * relayd command.
76 */
77void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
78{
79 int ret;
80 struct lttng_ht_iter iter;
81 struct consumer_socket *socket;
82
83 assert(consumer);
84
85 /* Destroy any relayd connection */
86 if (consumer && consumer->type == CONSUMER_DST_NET) {
87 rcu_read_lock();
88 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
89 node.node) {
90 /* Send destroy relayd command */
91 ret = consumer_send_destroy_relayd(socket, consumer);
92 if (ret < 0) {
93 ERR("Unable to send destroy relayd command to consumer");
94 /* Continue since we MUST delete everything at this point. */
95 }
96 }
97 rcu_read_unlock();
98 }
99}
100
a4b92340
DG
101/*
102 * From a consumer_data structure, allocate and add a consumer socket to the
103 * consumer output.
104 *
105 * Return 0 on success, else negative value on error
106 */
107int consumer_create_socket(struct consumer_data *data,
108 struct consumer_output *output)
109{
110 int ret = 0;
111 struct consumer_socket *socket;
112
113 assert(data);
114
115 if (output == NULL || data->cmd_sock < 0) {
116 /*
117 * Not an error. Possible there is simply not spawned consumer or it's
118 * disabled for the tracing session asking the socket.
119 */
120 goto error;
121 }
122
123 rcu_read_lock();
124 socket = consumer_find_socket(data->cmd_sock, output);
125 rcu_read_unlock();
126 if (socket == NULL) {
127 socket = consumer_allocate_socket(data->cmd_sock);
128 if (socket == NULL) {
129 ret = -1;
130 goto error;
131 }
132
2f77fc4b 133 socket->registered = 0;
a4b92340
DG
134 socket->lock = &data->lock;
135 rcu_read_lock();
136 consumer_add_socket(socket, output);
137 rcu_read_unlock();
138 }
139
140 DBG3("Consumer socket created (fd: %d) and added to output",
141 data->cmd_sock);
142
143error:
144 return ret;
145}
146
173af62f
DG
147/*
148 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
149 * be acquired before calling this function and across use of the
150 * returned consumer_socket.
151 */
152struct consumer_socket *consumer_find_socket(int key,
153 struct consumer_output *consumer)
154{
155 struct lttng_ht_iter iter;
156 struct lttng_ht_node_ulong *node;
157 struct consumer_socket *socket = NULL;
158
159 /* Negative keys are lookup failures */
a4b92340 160 if (key < 0 || consumer == NULL) {
173af62f
DG
161 return NULL;
162 }
163
164 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
165 &iter);
166 node = lttng_ht_iter_get_node_ulong(&iter);
167 if (node != NULL) {
168 socket = caa_container_of(node, struct consumer_socket, node);
169 }
170
171 return socket;
172}
173
174/*
175 * Allocate a new consumer_socket and return the pointer.
176 */
177struct consumer_socket *consumer_allocate_socket(int fd)
178{
179 struct consumer_socket *socket = NULL;
180
181 socket = zmalloc(sizeof(struct consumer_socket));
182 if (socket == NULL) {
183 PERROR("zmalloc consumer socket");
184 goto error;
185 }
186
187 socket->fd = fd;
188 lttng_ht_node_init_ulong(&socket->node, fd);
189
190error:
191 return socket;
192}
193
194/*
195 * Add consumer socket to consumer output object. Read side lock must be
196 * acquired before calling this function.
197 */
198void consumer_add_socket(struct consumer_socket *sock,
199 struct consumer_output *consumer)
200{
201 assert(sock);
202 assert(consumer);
203
204 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
205}
206
207/*
208 * Delte consumer socket to consumer output object. Read side lock must be
209 * acquired before calling this function.
210 */
211void consumer_del_socket(struct consumer_socket *sock,
212 struct consumer_output *consumer)
213{
214 int ret;
215 struct lttng_ht_iter iter;
216
217 assert(sock);
218 assert(consumer);
219
220 iter.iter.node = &sock->node.node;
221 ret = lttng_ht_del(consumer->socks, &iter);
222 assert(!ret);
223}
224
225/*
226 * RCU destroy call function.
227 */
228static void destroy_socket_rcu(struct rcu_head *head)
229{
230 struct lttng_ht_node_ulong *node =
231 caa_container_of(head, struct lttng_ht_node_ulong, head);
232 struct consumer_socket *socket =
233 caa_container_of(node, struct consumer_socket, node);
234
235 free(socket);
236}
237
238/*
239 * Destroy and free socket pointer in a call RCU. Read side lock must be
240 * acquired before calling this function.
241 */
242void consumer_destroy_socket(struct consumer_socket *sock)
243{
244 assert(sock);
245
246 /*
247 * We DO NOT close the file descriptor here since it is global to the
2f77fc4b
DG
248 * session daemon and is closed only if the consumer dies or a custom
249 * consumer was registered,
173af62f 250 */
2f77fc4b
DG
251 if (sock->registered) {
252 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
253 lttcomm_close_unix_sock(sock->fd);
254 }
173af62f
DG
255
256 call_rcu(&sock->node.head, destroy_socket_rcu);
257}
258
00e2e675
DG
259/*
260 * Allocate and assign data to a consumer_output object.
261 *
262 * Return pointer to structure.
263 */
264struct consumer_output *consumer_create_output(enum consumer_dst_type type)
265{
266 struct consumer_output *output = NULL;
267
268 output = zmalloc(sizeof(struct consumer_output));
269 if (output == NULL) {
270 PERROR("zmalloc consumer_output");
271 goto error;
272 }
273
274 /* By default, consumer output is enabled */
275 output->enabled = 1;
276 output->type = type;
277 output->net_seq_index = -1;
173af62f
DG
278
279 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675
DG
280
281error:
282 return output;
283}
284
285/*
286 * Delete the consumer_output object from the list and free the ptr.
287 */
288void consumer_destroy_output(struct consumer_output *obj)
289{
290 if (obj == NULL) {
291 return;
292 }
293
173af62f
DG
294 if (obj->socks) {
295 struct lttng_ht_iter iter;
296 struct consumer_socket *socket;
297
2f77fc4b 298 rcu_read_lock();
173af62f 299 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
2f77fc4b 300 consumer_del_socket(socket, obj);
173af62f
DG
301 consumer_destroy_socket(socket);
302 }
2f77fc4b
DG
303 rcu_read_unlock();
304
305 /* Finally destroy HT */
306 lttng_ht_destroy(obj->socks);
00e2e675 307 }
173af62f 308
00e2e675
DG
309 free(obj);
310}
311
312/*
313 * Copy consumer output and returned the newly allocated copy.
314 */
315struct consumer_output *consumer_copy_output(struct consumer_output *obj)
316{
173af62f
DG
317 struct lttng_ht_iter iter;
318 struct consumer_socket *socket, *copy_sock;
00e2e675
DG
319 struct consumer_output *output;
320
321 assert(obj);
322
323 output = consumer_create_output(obj->type);
324 if (output == NULL) {
325 goto error;
326 }
327
328 memcpy(output, obj, sizeof(struct consumer_output));
329
173af62f
DG
330 /* Copy sockets */
331 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
332
333 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
334 /* Create new socket object. */
335 copy_sock = consumer_allocate_socket(socket->fd);
336 if (copy_sock == NULL) {
337 goto malloc_error;
338 }
339
340 copy_sock->lock = socket->lock;
341 consumer_add_socket(copy_sock, output);
342 }
343
00e2e675
DG
344error:
345 return output;
173af62f
DG
346
347malloc_error:
348 consumer_destroy_output(output);
349 return NULL;
00e2e675
DG
350}
351
352/*
353 * Set network URI to the consumer output object.
354 *
ad20f474
DG
355 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
356 * error.
00e2e675
DG
357 */
358int consumer_set_network_uri(struct consumer_output *obj,
359 struct lttng_uri *uri)
360{
361 int ret;
362 char tmp_path[PATH_MAX];
363 char hostname[HOST_NAME_MAX];
364 struct lttng_uri *dst_uri = NULL;
365
366 /* Code flow error safety net. */
367 assert(obj);
368 assert(uri);
369
370 switch (uri->stype) {
371 case LTTNG_STREAM_CONTROL:
372 dst_uri = &obj->dst.net.control;
373 obj->dst.net.control_isset = 1;
374 if (uri->port == 0) {
375 /* Assign default port. */
376 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
377 }
ad20f474 378 DBG3("Consumer control URI set with port %d", uri->port);
00e2e675
DG
379 break;
380 case LTTNG_STREAM_DATA:
381 dst_uri = &obj->dst.net.data;
382 obj->dst.net.data_isset = 1;
383 if (uri->port == 0) {
384 /* Assign default port. */
385 uri->port = DEFAULT_NETWORK_DATA_PORT;
386 }
ad20f474 387 DBG3("Consumer data URI set with port %d", uri->port);
00e2e675
DG
388 break;
389 default:
390 ERR("Set network uri type unknown %d", uri->stype);
391 goto error;
392 }
393
394 ret = uri_compare(dst_uri, uri);
395 if (!ret) {
396 /* Same URI, don't touch it and return success. */
397 DBG3("URI network compare are the same");
ad20f474 398 goto equal;
00e2e675
DG
399 }
400
401 /* URIs were not equal, replacing it. */
402 memset(dst_uri, 0, sizeof(struct lttng_uri));
403 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
404 obj->type = CONSUMER_DST_NET;
405
406 /* Handle subdir and add hostname in front. */
407 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
408 /* Get hostname to append it in the pathname */
409 ret = gethostname(hostname, sizeof(hostname));
410 if (ret < 0) {
411 PERROR("gethostname. Fallback on default localhost");
412 strncpy(hostname, "localhost", sizeof(hostname));
413 }
414 hostname[sizeof(hostname) - 1] = '\0';
415
416 /* Setup consumer subdir if none present in the control URI */
417 if (strlen(dst_uri->subdir) == 0) {
418 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
419 hostname, obj->subdir);
420 } else {
421 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
422 hostname, dst_uri->subdir);
423 }
424 if (ret < 0) {
425 PERROR("snprintf set consumer uri subdir");
426 goto error;
427 }
428
429 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
430 DBG3("Consumer set network uri subdir path %s", tmp_path);
431 }
432
00e2e675 433 return 0;
ad20f474
DG
434equal:
435 return 1;
00e2e675
DG
436error:
437 return -1;
438}
439
440/*
441 * Send file descriptor to consumer via sock.
442 */
443int consumer_send_fds(int sock, int *fds, size_t nb_fd)
444{
445 int ret;
446
447 assert(fds);
448 assert(nb_fd > 0);
449
450 ret = lttcomm_send_fds_unix_sock(sock, fds, nb_fd);
451 if (ret < 0) {
452 PERROR("send consumer fds");
453 goto error;
454 }
455
456error:
457 return ret;
458}
459
460/*
461 * Consumer send channel communication message structure to consumer.
462 */
463int consumer_send_channel(int sock, struct lttcomm_consumer_msg *msg)
464{
465 int ret;
466
467 assert(msg);
468 assert(sock >= 0);
469
470 ret = lttcomm_send_unix_sock(sock, msg,
471 sizeof(struct lttcomm_consumer_msg));
472 if (ret < 0) {
473 PERROR("send consumer channel");
474 goto error;
475 }
476
477error:
478 return ret;
479}
480
481/*
482 * Init channel communication message structure.
483 */
484void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
485 enum lttng_consumer_command cmd,
486 int channel_key,
487 uint64_t max_sb_size,
488 uint64_t mmap_len,
c30aaa51
MD
489 const char *name,
490 unsigned int nb_init_streams)
00e2e675
DG
491{
492 assert(msg);
493
494 /* TODO: Args validation */
495
496 /* Zeroed structure */
497 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
498
499 /* Send channel */
500 msg->cmd_type = cmd;
501 msg->u.channel.channel_key = channel_key;
502 msg->u.channel.max_sb_size = max_sb_size;
503 msg->u.channel.mmap_len = mmap_len;
c30aaa51 504 msg->u.channel.nb_init_streams = nb_init_streams;
00e2e675
DG
505}
506
507/*
508 * Init stream communication message structure.
509 */
510void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
511 enum lttng_consumer_command cmd,
512 int channel_key,
513 int stream_key,
514 uint32_t state,
515 enum lttng_event_output output,
516 uint64_t mmap_len,
517 uid_t uid,
518 gid_t gid,
519 int net_index,
520 unsigned int metadata_flag,
521 const char *name,
ca22feea
DG
522 const char *pathname,
523 unsigned int session_id)
00e2e675
DG
524{
525 assert(msg);
526
527 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
528
529 /* TODO: Args validation */
530
531 msg->cmd_type = cmd;
532 msg->u.stream.channel_key = channel_key;
533 msg->u.stream.stream_key = stream_key;
534 msg->u.stream.state = state;
535 msg->u.stream.output = output;
536 msg->u.stream.mmap_len = mmap_len;
537 msg->u.stream.uid = uid;
538 msg->u.stream.gid = gid;
539 msg->u.stream.net_index = net_index;
540 msg->u.stream.metadata_flag = metadata_flag;
ca22feea 541 msg->u.stream.session_id = (uint64_t) session_id;
00e2e675
DG
542 strncpy(msg->u.stream.name, name, sizeof(msg->u.stream.name));
543 msg->u.stream.name[sizeof(msg->u.stream.name) - 1] = '\0';
544 strncpy(msg->u.stream.path_name, pathname,
545 sizeof(msg->u.stream.path_name));
546 msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
547}
548
549/*
550 * Send stream communication structure to the consumer.
551 */
552int consumer_send_stream(int sock, struct consumer_output *dst,
553 struct lttcomm_consumer_msg *msg, int *fds, size_t nb_fd)
554{
555 int ret;
556
557 assert(msg);
558 assert(dst);
559
560 switch (dst->type) {
561 case CONSUMER_DST_NET:
562 /* Consumer should send the stream on the network. */
563 msg->u.stream.net_index = dst->net_seq_index;
564 break;
565 case CONSUMER_DST_LOCAL:
566 /* Add stream file name to stream path */
c30ce0b3
CB
567 strncat(msg->u.stream.path_name, "/",
568 sizeof(msg->u.stream.path_name) -
569 strlen(msg->u.stream.path_name) - 1);
00e2e675 570 strncat(msg->u.stream.path_name, msg->u.stream.name,
c30ce0b3
CB
571 sizeof(msg->u.stream.path_name) -
572 strlen(msg->u.stream.path_name) - 1);
00e2e675
DG
573 msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
574 /* Indicate that the stream is NOT network */
575 msg->u.stream.net_index = -1;
576 break;
577 default:
578 ERR("Consumer unknown output type (%d)", dst->type);
579 ret = -1;
580 goto error;
581 }
582
583 /* Send on socket */
584 ret = lttcomm_send_unix_sock(sock, msg,
585 sizeof(struct lttcomm_consumer_msg));
586 if (ret < 0) {
587 PERROR("send consumer stream");
588 goto error;
589 }
590
591 ret = consumer_send_fds(sock, fds, nb_fd);
592 if (ret < 0) {
593 goto error;
594 }
595
596error:
597 return ret;
598}
37278a1e
DG
599
600/*
601 * Send relayd socket to consumer associated with a session name.
602 *
603 * On success return positive value. On error, negative value.
604 */
605int consumer_send_relayd_socket(int consumer_sock,
606 struct lttcomm_sock *sock, struct consumer_output *consumer,
607 enum lttng_stream_type type)
608{
609 int ret;
610 struct lttcomm_consumer_msg msg;
611
612 /* Code flow error. Safety net. */
613 assert(sock);
614 assert(consumer);
615
616 /* Bail out if consumer is disabled */
617 if (!consumer->enabled) {
f73fabfd 618 ret = LTTNG_OK;
37278a1e
DG
619 goto error;
620 }
621
622 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
623 /*
624 * Assign network consumer output index using the temporary consumer since
625 * this call should only be made from within a set_consumer_uri() function
626 * call in the session daemon.
627 */
628 msg.u.relayd_sock.net_index = consumer->net_seq_index;
629 msg.u.relayd_sock.type = type;
630 memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
631
173af62f 632 DBG3("Sending relayd sock info to consumer on %d", consumer_sock);
37278a1e
DG
633 ret = lttcomm_send_unix_sock(consumer_sock, &msg, sizeof(msg));
634 if (ret < 0) {
635 PERROR("send consumer relayd socket info");
636 goto error;
637 }
638
639 DBG3("Sending relayd socket file descriptor to consumer");
640 ret = consumer_send_fds(consumer_sock, &sock->fd, 1);
641 if (ret < 0) {
642 goto error;
643 }
644
645 DBG2("Consumer relayd socket sent");
646
647error:
648 return ret;
649}
173af62f
DG
650
651/*
2f77fc4b
DG
652 * Set consumer subdirectory using the session name and a generated datetime if
653 * needed. This is appended to the current subdirectory.
173af62f 654 */
2f77fc4b
DG
655int consumer_set_subdir(struct consumer_output *consumer,
656 const char *session_name)
173af62f 657{
2f77fc4b
DG
658 int ret = 0;
659 unsigned int have_default_name = 0;
660 char datetime[16], tmp_path[PATH_MAX];
661 time_t rawtime;
662 struct tm *timeinfo;
173af62f
DG
663
664 assert(consumer);
2f77fc4b
DG
665 assert(session_name);
666
667 memset(tmp_path, 0, sizeof(tmp_path));
668
669 /* Flag if we have a default session. */
670 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
671 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
672 have_default_name = 1;
673 } else {
674 /* Get date and time for session path */
675 time(&rawtime);
676 timeinfo = localtime(&rawtime);
677 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
173af62f
DG
678 }
679
2f77fc4b
DG
680 if (have_default_name) {
681 ret = snprintf(tmp_path, sizeof(tmp_path),
682 "%s/%s", consumer->subdir, session_name);
683 } else {
684 ret = snprintf(tmp_path, sizeof(tmp_path),
685 "%s/%s-%s/", consumer->subdir, session_name, datetime);
686 }
173af62f 687 if (ret < 0) {
2f77fc4b 688 PERROR("snprintf session name date");
173af62f
DG
689 goto error;
690 }
691
2f77fc4b
DG
692 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
693 DBG2("Consumer subdir set to %s", consumer->subdir);
173af62f
DG
694
695error:
696 return ret;
697}
This page took 0.083091 seconds and 4 git commands to generate.