Add lttng-error.h containing every API err. code
[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,
489 const char *name)
490{
491 assert(msg);
492
493 /* TODO: Args validation */
494
495 /* Zeroed structure */
496 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
497
498 /* Send channel */
499 msg->cmd_type = cmd;
500 msg->u.channel.channel_key = channel_key;
501 msg->u.channel.max_sb_size = max_sb_size;
502 msg->u.channel.mmap_len = mmap_len;
503}
504
505/*
506 * Init stream communication message structure.
507 */
508void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
509 enum lttng_consumer_command cmd,
510 int channel_key,
511 int stream_key,
512 uint32_t state,
513 enum lttng_event_output output,
514 uint64_t mmap_len,
515 uid_t uid,
516 gid_t gid,
517 int net_index,
518 unsigned int metadata_flag,
519 const char *name,
520 const char *pathname)
521{
522 assert(msg);
523
524 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
525
526 /* TODO: Args validation */
527
528 msg->cmd_type = cmd;
529 msg->u.stream.channel_key = channel_key;
530 msg->u.stream.stream_key = stream_key;
531 msg->u.stream.state = state;
532 msg->u.stream.output = output;
533 msg->u.stream.mmap_len = mmap_len;
534 msg->u.stream.uid = uid;
535 msg->u.stream.gid = gid;
536 msg->u.stream.net_index = net_index;
537 msg->u.stream.metadata_flag = metadata_flag;
538 strncpy(msg->u.stream.name, name, sizeof(msg->u.stream.name));
539 msg->u.stream.name[sizeof(msg->u.stream.name) - 1] = '\0';
540 strncpy(msg->u.stream.path_name, pathname,
541 sizeof(msg->u.stream.path_name));
542 msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
543}
544
545/*
546 * Send stream communication structure to the consumer.
547 */
548int consumer_send_stream(int sock, struct consumer_output *dst,
549 struct lttcomm_consumer_msg *msg, int *fds, size_t nb_fd)
550{
551 int ret;
552
553 assert(msg);
554 assert(dst);
555
556 switch (dst->type) {
557 case CONSUMER_DST_NET:
558 /* Consumer should send the stream on the network. */
559 msg->u.stream.net_index = dst->net_seq_index;
560 break;
561 case CONSUMER_DST_LOCAL:
562 /* Add stream file name to stream path */
c30ce0b3
CB
563 strncat(msg->u.stream.path_name, "/",
564 sizeof(msg->u.stream.path_name) -
565 strlen(msg->u.stream.path_name) - 1);
00e2e675 566 strncat(msg->u.stream.path_name, msg->u.stream.name,
c30ce0b3
CB
567 sizeof(msg->u.stream.path_name) -
568 strlen(msg->u.stream.path_name) - 1);
00e2e675
DG
569 msg->u.stream.path_name[sizeof(msg->u.stream.path_name) - 1] = '\0';
570 /* Indicate that the stream is NOT network */
571 msg->u.stream.net_index = -1;
572 break;
573 default:
574 ERR("Consumer unknown output type (%d)", dst->type);
575 ret = -1;
576 goto error;
577 }
578
579 /* Send on socket */
580 ret = lttcomm_send_unix_sock(sock, msg,
581 sizeof(struct lttcomm_consumer_msg));
582 if (ret < 0) {
583 PERROR("send consumer stream");
584 goto error;
585 }
586
587 ret = consumer_send_fds(sock, fds, nb_fd);
588 if (ret < 0) {
589 goto error;
590 }
591
592error:
593 return ret;
594}
37278a1e
DG
595
596/*
597 * Send relayd socket to consumer associated with a session name.
598 *
599 * On success return positive value. On error, negative value.
600 */
601int consumer_send_relayd_socket(int consumer_sock,
602 struct lttcomm_sock *sock, struct consumer_output *consumer,
603 enum lttng_stream_type type)
604{
605 int ret;
606 struct lttcomm_consumer_msg msg;
607
608 /* Code flow error. Safety net. */
609 assert(sock);
610 assert(consumer);
611
612 /* Bail out if consumer is disabled */
613 if (!consumer->enabled) {
f73fabfd 614 ret = LTTNG_OK;
37278a1e
DG
615 goto error;
616 }
617
618 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
619 /*
620 * Assign network consumer output index using the temporary consumer since
621 * this call should only be made from within a set_consumer_uri() function
622 * call in the session daemon.
623 */
624 msg.u.relayd_sock.net_index = consumer->net_seq_index;
625 msg.u.relayd_sock.type = type;
626 memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
627
173af62f 628 DBG3("Sending relayd sock info to consumer on %d", consumer_sock);
37278a1e
DG
629 ret = lttcomm_send_unix_sock(consumer_sock, &msg, sizeof(msg));
630 if (ret < 0) {
631 PERROR("send consumer relayd socket info");
632 goto error;
633 }
634
635 DBG3("Sending relayd socket file descriptor to consumer");
636 ret = consumer_send_fds(consumer_sock, &sock->fd, 1);
637 if (ret < 0) {
638 goto error;
639 }
640
641 DBG2("Consumer relayd socket sent");
642
643error:
644 return ret;
645}
173af62f
DG
646
647/*
2f77fc4b
DG
648 * Set consumer subdirectory using the session name and a generated datetime if
649 * needed. This is appended to the current subdirectory.
173af62f 650 */
2f77fc4b
DG
651int consumer_set_subdir(struct consumer_output *consumer,
652 const char *session_name)
173af62f 653{
2f77fc4b
DG
654 int ret = 0;
655 unsigned int have_default_name = 0;
656 char datetime[16], tmp_path[PATH_MAX];
657 time_t rawtime;
658 struct tm *timeinfo;
173af62f
DG
659
660 assert(consumer);
2f77fc4b
DG
661 assert(session_name);
662
663 memset(tmp_path, 0, sizeof(tmp_path));
664
665 /* Flag if we have a default session. */
666 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
667 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
668 have_default_name = 1;
669 } else {
670 /* Get date and time for session path */
671 time(&rawtime);
672 timeinfo = localtime(&rawtime);
673 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
173af62f
DG
674 }
675
2f77fc4b
DG
676 if (have_default_name) {
677 ret = snprintf(tmp_path, sizeof(tmp_path),
678 "%s/%s", consumer->subdir, session_name);
679 } else {
680 ret = snprintf(tmp_path, sizeof(tmp_path),
681 "%s/%s-%s/", consumer->subdir, session_name, datetime);
682 }
173af62f 683 if (ret < 0) {
2f77fc4b 684 PERROR("snprintf session name date");
173af62f
DG
685 goto error;
686 }
687
2f77fc4b
DG
688 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
689 DBG2("Consumer subdir set to %s", consumer->subdir);
173af62f
DG
690
691error:
692 return ret;
693}
This page took 0.079508 seconds and 4 git commands to generate.