relayd: Add backward compatibility for --group-output-by-session
[lttng-tools.git] / src / bin / lttng-relayd / main.c
1 /*
2 * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 * 2013 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 *
7 * This program is free software; you can redistribute it and/or modify
8 * it under the terms of the GNU General Public License, version 2 only,
9 * as published by the Free Software Foundation.
10 *
11 * This program is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 * more details.
15 *
16 * You should have received a copy of the GNU General Public License along
17 * with this program; if not, write to the Free Software Foundation, Inc.,
18 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19 */
20
21 #define _LGPL_SOURCE
22 #include <getopt.h>
23 #include <grp.h>
24 #include <limits.h>
25 #include <pthread.h>
26 #include <signal.h>
27 #include <stdio.h>
28 #include <stdlib.h>
29 #include <string.h>
30 #include <sys/mman.h>
31 #include <sys/mount.h>
32 #include <sys/resource.h>
33 #include <sys/socket.h>
34 #include <sys/stat.h>
35 #include <sys/types.h>
36 #include <sys/wait.h>
37 #include <inttypes.h>
38 #include <urcu/futex.h>
39 #include <urcu/uatomic.h>
40 #include <urcu/rculist.h>
41 #include <unistd.h>
42 #include <fcntl.h>
43 #include <strings.h>
44
45 #include <lttng/lttng.h>
46 #include <common/common.h>
47 #include <common/compat/poll.h>
48 #include <common/compat/socket.h>
49 #include <common/compat/endian.h>
50 #include <common/compat/getenv.h>
51 #include <common/defaults.h>
52 #include <common/daemonize.h>
53 #include <common/futex.h>
54 #include <common/sessiond-comm/sessiond-comm.h>
55 #include <common/sessiond-comm/inet.h>
56 #include <common/sessiond-comm/relayd.h>
57 #include <common/uri.h>
58 #include <common/utils.h>
59 #include <common/align.h>
60 #include <common/config/session-config.h>
61 #include <common/dynamic-buffer.h>
62 #include <common/buffer-view.h>
63 #include <common/string-utils/format.h>
64
65 #include "backward-compatibility-group-by.h"
66 #include "cmd.h"
67 #include "connection.h"
68 #include "ctf-trace.h"
69 #include "health-relayd.h"
70 #include "index.h"
71 #include "live.h"
72 #include "lttng-relayd.h"
73 #include "session.h"
74 #include "sessiond-trace-chunks.h"
75 #include "stream.h"
76 #include "tcp_keep_alive.h"
77 #include "testpoint.h"
78 #include "tracefile-array.h"
79 #include "utils.h"
80 #include "version.h"
81 #include "viewer-stream.h"
82
83 static const char *help_msg =
84 #ifdef LTTNG_EMBED_HELP
85 #include <lttng-relayd.8.h>
86 #else
87 NULL
88 #endif
89 ;
90
91 enum relay_connection_status {
92 RELAY_CONNECTION_STATUS_OK,
93 /* An error occurred while processing an event on the connection. */
94 RELAY_CONNECTION_STATUS_ERROR,
95 /* Connection closed/shutdown cleanly. */
96 RELAY_CONNECTION_STATUS_CLOSED,
97 };
98
99 /* command line options */
100 char *opt_output_path, *opt_working_directory;
101 static int opt_daemon, opt_background, opt_print_version;
102 enum relay_group_output_by opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_UNKNOWN;
103
104 /*
105 * We need to wait for listener and live listener threads, as well as
106 * health check thread, before being ready to signal readiness.
107 */
108 #define NR_LTTNG_RELAY_READY 3
109 static int lttng_relay_ready = NR_LTTNG_RELAY_READY;
110
111 /* Size of receive buffer. */
112 #define RECV_DATA_BUFFER_SIZE 65536
113
114 static int recv_child_signal; /* Set to 1 when a SIGUSR1 signal is received. */
115 static pid_t child_ppid; /* Internal parent PID use with daemonize. */
116
117 static struct lttng_uri *control_uri;
118 static struct lttng_uri *data_uri;
119 static struct lttng_uri *live_uri;
120
121 const char *progname;
122
123 const char *tracing_group_name = DEFAULT_TRACING_GROUP;
124 static int tracing_group_name_override;
125
126 const char * const config_section_name = "relayd";
127
128 /*
129 * Quit pipe for all threads. This permits a single cancellation point
130 * for all threads when receiving an event on the pipe.
131 */
132 int thread_quit_pipe[2] = { -1, -1 };
133
134 /*
135 * This pipe is used to inform the worker thread that a command is queued and
136 * ready to be processed.
137 */
138 static int relay_conn_pipe[2] = { -1, -1 };
139
140 /* Shared between threads */
141 static int dispatch_thread_exit;
142
143 static pthread_t listener_thread;
144 static pthread_t dispatcher_thread;
145 static pthread_t worker_thread;
146 static pthread_t health_thread;
147
148 /*
149 * last_relay_stream_id_lock protects last_relay_stream_id increment
150 * atomicity on 32-bit architectures.
151 */
152 static pthread_mutex_t last_relay_stream_id_lock = PTHREAD_MUTEX_INITIALIZER;
153 static uint64_t last_relay_stream_id;
154
155 /*
156 * Relay command queue.
157 *
158 * The relay_thread_listener and relay_thread_dispatcher communicate with this
159 * queue.
160 */
161 static struct relay_conn_queue relay_conn_queue;
162
163 /* Global relay stream hash table. */
164 struct lttng_ht *relay_streams_ht;
165
166 /* Global relay viewer stream hash table. */
167 struct lttng_ht *viewer_streams_ht;
168
169 /* Global relay sessions hash table. */
170 struct lttng_ht *sessions_ht;
171
172 /* Relayd health monitoring */
173 struct health_app *health_relayd;
174
175 struct sessiond_trace_chunk_registry *sessiond_trace_chunk_registry;
176
177 static struct option long_options[] = {
178 { "control-port", 1, 0, 'C', },
179 { "data-port", 1, 0, 'D', },
180 { "live-port", 1, 0, 'L', },
181 { "daemonize", 0, 0, 'd', },
182 { "background", 0, 0, 'b', },
183 { "group", 1, 0, 'g', },
184 { "help", 0, 0, 'h', },
185 { "output", 1, 0, 'o', },
186 { "verbose", 0, 0, 'v', },
187 { "config", 1, 0, 'f' },
188 { "version", 0, 0, 'V' },
189 { "working-directory", 1, 0, 'w', },
190 { "group-output-by-session", 0, 0, 's', },
191 { "group-output-by-host", 0, 0, 'p', },
192 { NULL, 0, 0, 0, },
193 };
194
195 static const char *config_ignore_options[] = { "help", "config", "version" };
196
197 static void print_version(void) {
198 fprintf(stdout, "%s\n", VERSION);
199 }
200
201 static void relayd_config_log(void)
202 {
203 DBG("LTTng-relayd " VERSION " - " VERSION_NAME "%s%s",
204 GIT_VERSION[0] == '\0' ? "" : " - " GIT_VERSION,
205 EXTRA_VERSION_NAME[0] == '\0' ? "" : " - " EXTRA_VERSION_NAME);
206 if (EXTRA_VERSION_DESCRIPTION[0] != '\0') {
207 DBG("LTTng-relayd extra version description:\n\t" EXTRA_VERSION_DESCRIPTION "\n");
208 }
209 if (EXTRA_VERSION_PATCHES[0] != '\0') {
210 DBG("LTTng-relayd extra patches:\n\t" EXTRA_VERSION_PATCHES "\n");
211 }
212 }
213
214 /*
215 * Take an option from the getopt output and set it in the right variable to be
216 * used later.
217 *
218 * Return 0 on success else a negative value.
219 */
220 static int set_option(int opt, const char *arg, const char *optname)
221 {
222 int ret;
223
224 switch (opt) {
225 case 0:
226 fprintf(stderr, "option %s", optname);
227 if (arg) {
228 fprintf(stderr, " with arg %s\n", arg);
229 }
230 break;
231 case 'C':
232 if (lttng_is_setuid_setgid()) {
233 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
234 "-C, --control-port");
235 } else {
236 ret = uri_parse(arg, &control_uri);
237 if (ret < 0) {
238 ERR("Invalid control URI specified");
239 goto end;
240 }
241 if (control_uri->port == 0) {
242 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
243 }
244 }
245 break;
246 case 'D':
247 if (lttng_is_setuid_setgid()) {
248 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
249 "-D, -data-port");
250 } else {
251 ret = uri_parse(arg, &data_uri);
252 if (ret < 0) {
253 ERR("Invalid data URI specified");
254 goto end;
255 }
256 if (data_uri->port == 0) {
257 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
258 }
259 }
260 break;
261 case 'L':
262 if (lttng_is_setuid_setgid()) {
263 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
264 "-L, -live-port");
265 } else {
266 ret = uri_parse(arg, &live_uri);
267 if (ret < 0) {
268 ERR("Invalid live URI specified");
269 goto end;
270 }
271 if (live_uri->port == 0) {
272 live_uri->port = DEFAULT_NETWORK_VIEWER_PORT;
273 }
274 }
275 break;
276 case 'd':
277 opt_daemon = 1;
278 break;
279 case 'b':
280 opt_background = 1;
281 break;
282 case 'g':
283 if (lttng_is_setuid_setgid()) {
284 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
285 "-g, --group");
286 } else {
287 tracing_group_name = strdup(arg);
288 if (tracing_group_name == NULL) {
289 ret = -errno;
290 PERROR("strdup");
291 goto end;
292 }
293 tracing_group_name_override = 1;
294 }
295 break;
296 case 'h':
297 ret = utils_show_help(8, "lttng-relayd", help_msg);
298 if (ret) {
299 ERR("Cannot show --help for `lttng-relayd`");
300 perror("exec");
301 }
302 exit(EXIT_FAILURE);
303 case 'V':
304 opt_print_version = 1;
305 break;
306 case 'o':
307 if (lttng_is_setuid_setgid()) {
308 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
309 "-o, --output");
310 } else {
311 ret = asprintf(&opt_output_path, "%s", arg);
312 if (ret < 0) {
313 ret = -errno;
314 PERROR("asprintf opt_output_path");
315 goto end;
316 }
317 }
318 break;
319 case 'w':
320 if (lttng_is_setuid_setgid()) {
321 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
322 "-w, --working-directory");
323 } else {
324 ret = asprintf(&opt_working_directory, "%s", arg);
325 if (ret < 0) {
326 ret = -errno;
327 PERROR("asprintf opt_working_directory");
328 goto end;
329 }
330 }
331 break;
332
333 case 'v':
334 /* Verbose level can increase using multiple -v */
335 if (arg) {
336 lttng_opt_verbose = config_parse_value(arg);
337 } else {
338 /* Only 3 level of verbosity (-vvv). */
339 if (lttng_opt_verbose < 3) {
340 lttng_opt_verbose += 1;
341 }
342 }
343 break;
344 case 's':
345 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
346 ERR("Cannot set --group-output-by-session, another --group-output-by argument is present");
347 exit(EXIT_FAILURE);
348 }
349 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_SESSION;
350 break;
351 case 'p':
352 if (opt_group_output_by != RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
353 ERR("Cannot set --group-output-by-host, another --group-output-by argument is present");
354 exit(EXIT_FAILURE);
355 }
356 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
357 break;
358 default:
359 /* Unknown option or other error.
360 * Error is printed by getopt, just return */
361 ret = -1;
362 goto end;
363 }
364
365 /* All good. */
366 ret = 0;
367
368 end:
369 return ret;
370 }
371
372 /*
373 * config_entry_handler_cb used to handle options read from a config file.
374 * See config_entry_handler_cb comment in common/config/session-config.h for the
375 * return value conventions.
376 */
377 static int config_entry_handler(const struct config_entry *entry, void *unused)
378 {
379 int ret = 0, i;
380
381 if (!entry || !entry->name || !entry->value) {
382 ret = -EINVAL;
383 goto end;
384 }
385
386 /* Check if the option is to be ignored */
387 for (i = 0; i < sizeof(config_ignore_options) / sizeof(char *); i++) {
388 if (!strcmp(entry->name, config_ignore_options[i])) {
389 goto end;
390 }
391 }
392
393 for (i = 0; i < (sizeof(long_options) / sizeof(struct option)) - 1; i++) {
394 /* Ignore if entry name is not fully matched. */
395 if (strcmp(entry->name, long_options[i].name)) {
396 continue;
397 }
398
399 /*
400 * If the option takes no argument on the command line,
401 * we have to check if the value is "true". We support
402 * non-zero numeric values, true, on and yes.
403 */
404 if (!long_options[i].has_arg) {
405 ret = config_parse_value(entry->value);
406 if (ret <= 0) {
407 if (ret) {
408 WARN("Invalid configuration value \"%s\" for option %s",
409 entry->value, entry->name);
410 }
411 /* False, skip boolean config option. */
412 goto end;
413 }
414 }
415
416 ret = set_option(long_options[i].val, entry->value, entry->name);
417 goto end;
418 }
419
420 WARN("Unrecognized option \"%s\" in daemon configuration file.",
421 entry->name);
422
423 end:
424 return ret;
425 }
426
427 static int parse_env_options(void)
428 {
429 int ret = 0;
430 char *value = NULL;
431
432 value = lttng_secure_getenv(DEFAULT_LTTNG_RELAYD_WORKING_DIRECTORY_ENV);
433 if (value) {
434 opt_working_directory = strdup(value);
435 if (!opt_working_directory) {
436 ERR("Failed to allocate working directory string (\"%s\")",
437 value);
438 ret = -1;
439 }
440 }
441 return ret;
442 }
443
444 static int set_options(int argc, char **argv)
445 {
446 int c, ret = 0, option_index = 0, retval = 0;
447 int orig_optopt = optopt, orig_optind = optind;
448 char *default_address, *optstring;
449 const char *config_path = NULL;
450
451 optstring = utils_generate_optstring(long_options,
452 sizeof(long_options) / sizeof(struct option));
453 if (!optstring) {
454 retval = -ENOMEM;
455 goto exit;
456 }
457
458 /* Check for the --config option */
459
460 while ((c = getopt_long(argc, argv, optstring, long_options,
461 &option_index)) != -1) {
462 if (c == '?') {
463 retval = -EINVAL;
464 goto exit;
465 } else if (c != 'f') {
466 continue;
467 }
468
469 if (lttng_is_setuid_setgid()) {
470 WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
471 "-f, --config");
472 } else {
473 config_path = utils_expand_path(optarg);
474 if (!config_path) {
475 ERR("Failed to resolve path: %s", optarg);
476 }
477 }
478 }
479
480 ret = config_get_section_entries(config_path, config_section_name,
481 config_entry_handler, NULL);
482 if (ret) {
483 if (ret > 0) {
484 ERR("Invalid configuration option at line %i", ret);
485 }
486 retval = -1;
487 goto exit;
488 }
489
490 /* Reset getopt's global state */
491 optopt = orig_optopt;
492 optind = orig_optind;
493 while (1) {
494 c = getopt_long(argc, argv, optstring, long_options, &option_index);
495 if (c == -1) {
496 break;
497 }
498
499 ret = set_option(c, optarg, long_options[option_index].name);
500 if (ret < 0) {
501 retval = -1;
502 goto exit;
503 }
504 }
505
506 /* assign default values */
507 if (control_uri == NULL) {
508 ret = asprintf(&default_address,
509 "tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
510 DEFAULT_NETWORK_CONTROL_PORT);
511 if (ret < 0) {
512 PERROR("asprintf default data address");
513 retval = -1;
514 goto exit;
515 }
516
517 ret = uri_parse(default_address, &control_uri);
518 free(default_address);
519 if (ret < 0) {
520 ERR("Invalid control URI specified");
521 retval = -1;
522 goto exit;
523 }
524 }
525 if (data_uri == NULL) {
526 ret = asprintf(&default_address,
527 "tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
528 DEFAULT_NETWORK_DATA_PORT);
529 if (ret < 0) {
530 PERROR("asprintf default data address");
531 retval = -1;
532 goto exit;
533 }
534
535 ret = uri_parse(default_address, &data_uri);
536 free(default_address);
537 if (ret < 0) {
538 ERR("Invalid data URI specified");
539 retval = -1;
540 goto exit;
541 }
542 }
543 if (live_uri == NULL) {
544 ret = asprintf(&default_address,
545 "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
546 DEFAULT_NETWORK_VIEWER_PORT);
547 if (ret < 0) {
548 PERROR("asprintf default viewer control address");
549 retval = -1;
550 goto exit;
551 }
552
553 ret = uri_parse(default_address, &live_uri);
554 free(default_address);
555 if (ret < 0) {
556 ERR("Invalid viewer control URI specified");
557 retval = -1;
558 goto exit;
559 }
560 }
561
562 if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_UNKNOWN) {
563 opt_group_output_by = RELAYD_GROUP_OUTPUT_BY_HOST;
564 }
565
566 exit:
567 free(optstring);
568 return retval;
569 }
570
571 static void print_global_objects(void)
572 {
573 rcu_register_thread();
574
575 print_viewer_streams();
576 print_relay_streams();
577 print_sessions();
578
579 rcu_unregister_thread();
580 }
581
582 /*
583 * Cleanup the daemon
584 */
585 static void relayd_cleanup(void)
586 {
587 print_global_objects();
588
589 DBG("Cleaning up");
590
591 if (viewer_streams_ht)
592 lttng_ht_destroy(viewer_streams_ht);
593 if (relay_streams_ht)
594 lttng_ht_destroy(relay_streams_ht);
595 if (sessions_ht)
596 lttng_ht_destroy(sessions_ht);
597
598 free(opt_output_path);
599 free(opt_working_directory);
600
601 /* Close thread quit pipes */
602 utils_close_pipe(thread_quit_pipe);
603
604 uri_free(control_uri);
605 uri_free(data_uri);
606 /* Live URI is freed in the live thread. */
607
608 if (tracing_group_name_override) {
609 free((void *) tracing_group_name);
610 }
611 }
612
613 /*
614 * Write to writable pipe used to notify a thread.
615 */
616 static int notify_thread_pipe(int wpipe)
617 {
618 ssize_t ret;
619
620 ret = lttng_write(wpipe, "!", 1);
621 if (ret < 1) {
622 PERROR("write poll pipe");
623 goto end;
624 }
625 ret = 0;
626 end:
627 return ret;
628 }
629
630 static int notify_health_quit_pipe(int *pipe)
631 {
632 ssize_t ret;
633
634 ret = lttng_write(pipe[1], "4", 1);
635 if (ret < 1) {
636 PERROR("write relay health quit");
637 goto end;
638 }
639 ret = 0;
640 end:
641 return ret;
642 }
643
644 /*
645 * Stop all relayd and relayd-live threads.
646 */
647 int lttng_relay_stop_threads(void)
648 {
649 int retval = 0;
650
651 /* Stopping all threads */
652 DBG("Terminating all threads");
653 if (notify_thread_pipe(thread_quit_pipe[1])) {
654 ERR("write error on thread quit pipe");
655 retval = -1;
656 }
657
658 if (notify_health_quit_pipe(health_quit_pipe)) {
659 ERR("write error on health quit pipe");
660 }
661
662 /* Dispatch thread */
663 CMM_STORE_SHARED(dispatch_thread_exit, 1);
664 futex_nto1_wake(&relay_conn_queue.futex);
665
666 if (relayd_live_stop()) {
667 ERR("Error stopping live threads");
668 retval = -1;
669 }
670 return retval;
671 }
672
673 /*
674 * Signal handler for the daemon
675 *
676 * Simply stop all worker threads, leaving main() return gracefully after
677 * joining all threads and calling cleanup().
678 */
679 static void sighandler(int sig)
680 {
681 switch (sig) {
682 case SIGINT:
683 DBG("SIGINT caught");
684 if (lttng_relay_stop_threads()) {
685 ERR("Error stopping threads");
686 }
687 break;
688 case SIGTERM:
689 DBG("SIGTERM caught");
690 if (lttng_relay_stop_threads()) {
691 ERR("Error stopping threads");
692 }
693 break;
694 case SIGUSR1:
695 CMM_STORE_SHARED(recv_child_signal, 1);
696 break;
697 default:
698 break;
699 }
700 }
701
702 /*
703 * Setup signal handler for :
704 * SIGINT, SIGTERM, SIGPIPE
705 */
706 static int set_signal_handler(void)
707 {
708 int ret = 0;
709 struct sigaction sa;
710 sigset_t sigset;
711
712 if ((ret = sigemptyset(&sigset)) < 0) {
713 PERROR("sigemptyset");
714 return ret;
715 }
716
717 sa.sa_mask = sigset;
718 sa.sa_flags = 0;
719
720 sa.sa_handler = sighandler;
721 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
722 PERROR("sigaction");
723 return ret;
724 }
725
726 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
727 PERROR("sigaction");
728 return ret;
729 }
730
731 if ((ret = sigaction(SIGUSR1, &sa, NULL)) < 0) {
732 PERROR("sigaction");
733 return ret;
734 }
735
736 sa.sa_handler = SIG_IGN;
737 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
738 PERROR("sigaction");
739 return ret;
740 }
741
742 DBG("Signal handler set for SIGTERM, SIGUSR1, SIGPIPE and SIGINT");
743
744 return ret;
745 }
746
747 void lttng_relay_notify_ready(void)
748 {
749 /* Notify the parent of the fork() process that we are ready. */
750 if (opt_daemon || opt_background) {
751 if (uatomic_sub_return(&lttng_relay_ready, 1) == 0) {
752 kill(child_ppid, SIGUSR1);
753 }
754 }
755 }
756
757 /*
758 * Init thread quit pipe.
759 *
760 * Return -1 on error or 0 if all pipes are created.
761 */
762 static int init_thread_quit_pipe(void)
763 {
764 int ret;
765
766 ret = utils_create_pipe_cloexec(thread_quit_pipe);
767
768 return ret;
769 }
770
771 /*
772 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
773 */
774 static int create_thread_poll_set(struct lttng_poll_event *events, int size)
775 {
776 int ret;
777
778 if (events == NULL || size == 0) {
779 ret = -1;
780 goto error;
781 }
782
783 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
784 if (ret < 0) {
785 goto error;
786 }
787
788 /* Add quit pipe */
789 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
790 if (ret < 0) {
791 goto error;
792 }
793
794 return 0;
795
796 error:
797 return ret;
798 }
799
800 /*
801 * Check if the thread quit pipe was triggered.
802 *
803 * Return 1 if it was triggered else 0;
804 */
805 static int check_thread_quit_pipe(int fd, uint32_t events)
806 {
807 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
808 return 1;
809 }
810
811 return 0;
812 }
813
814 /*
815 * Create and init socket from uri.
816 */
817 static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri)
818 {
819 int ret;
820 struct lttcomm_sock *sock = NULL;
821
822 sock = lttcomm_alloc_sock_from_uri(uri);
823 if (sock == NULL) {
824 ERR("Allocating socket");
825 goto error;
826 }
827
828 ret = lttcomm_create_sock(sock);
829 if (ret < 0) {
830 goto error;
831 }
832 DBG("Listening on sock %d", sock->fd);
833
834 ret = sock->ops->bind(sock);
835 if (ret < 0) {
836 PERROR("Failed to bind socket");
837 goto error;
838 }
839
840 ret = sock->ops->listen(sock, -1);
841 if (ret < 0) {
842 goto error;
843
844 }
845
846 return sock;
847
848 error:
849 if (sock) {
850 lttcomm_destroy_sock(sock);
851 }
852 return NULL;
853 }
854
855 /*
856 * This thread manages the listening for new connections on the network
857 */
858 static void *relay_thread_listener(void *data)
859 {
860 int i, ret, pollfd, err = -1;
861 uint32_t revents, nb_fd;
862 struct lttng_poll_event events;
863 struct lttcomm_sock *control_sock, *data_sock;
864
865 DBG("[thread] Relay listener started");
866
867 health_register(health_relayd, HEALTH_RELAYD_TYPE_LISTENER);
868
869 health_code_update();
870
871 control_sock = relay_socket_create(control_uri);
872 if (!control_sock) {
873 goto error_sock_control;
874 }
875
876 data_sock = relay_socket_create(data_uri);
877 if (!data_sock) {
878 goto error_sock_relay;
879 }
880
881 /*
882 * Pass 3 as size here for the thread quit pipe, control and
883 * data socket.
884 */
885 ret = create_thread_poll_set(&events, 3);
886 if (ret < 0) {
887 goto error_create_poll;
888 }
889
890 /* Add the control socket */
891 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
892 if (ret < 0) {
893 goto error_poll_add;
894 }
895
896 /* Add the data socket */
897 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
898 if (ret < 0) {
899 goto error_poll_add;
900 }
901
902 lttng_relay_notify_ready();
903
904 if (testpoint(relayd_thread_listener)) {
905 goto error_testpoint;
906 }
907
908 while (1) {
909 health_code_update();
910
911 DBG("Listener accepting connections");
912
913 restart:
914 health_poll_entry();
915 ret = lttng_poll_wait(&events, -1);
916 health_poll_exit();
917 if (ret < 0) {
918 /*
919 * Restart interrupted system call.
920 */
921 if (errno == EINTR) {
922 goto restart;
923 }
924 goto error;
925 }
926
927 nb_fd = ret;
928
929 DBG("Relay new connection received");
930 for (i = 0; i < nb_fd; i++) {
931 health_code_update();
932
933 /* Fetch once the poll data */
934 revents = LTTNG_POLL_GETEV(&events, i);
935 pollfd = LTTNG_POLL_GETFD(&events, i);
936
937 /* Thread quit pipe has been closed. Killing thread. */
938 ret = check_thread_quit_pipe(pollfd, revents);
939 if (ret) {
940 err = 0;
941 goto exit;
942 }
943
944 if (revents & LPOLLIN) {
945 /*
946 * A new connection is requested, therefore a
947 * sessiond/consumerd connection is allocated in
948 * this thread, enqueued to a global queue and
949 * dequeued (and freed) in the worker thread.
950 */
951 int val = 1;
952 struct relay_connection *new_conn;
953 struct lttcomm_sock *newsock;
954 enum connection_type type;
955
956 if (pollfd == data_sock->fd) {
957 type = RELAY_DATA;
958 newsock = data_sock->ops->accept(data_sock);
959 DBG("Relay data connection accepted, socket %d",
960 newsock->fd);
961 } else {
962 assert(pollfd == control_sock->fd);
963 type = RELAY_CONTROL;
964 newsock = control_sock->ops->accept(control_sock);
965 DBG("Relay control connection accepted, socket %d",
966 newsock->fd);
967 }
968 if (!newsock) {
969 PERROR("accepting sock");
970 goto error;
971 }
972
973 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR, &val,
974 sizeof(val));
975 if (ret < 0) {
976 PERROR("setsockopt inet");
977 lttcomm_destroy_sock(newsock);
978 goto error;
979 }
980
981 ret = socket_apply_keep_alive_config(newsock->fd);
982 if (ret < 0) {
983 ERR("Failed to apply TCP keep-alive configuration on socket (%i)",
984 newsock->fd);
985 lttcomm_destroy_sock(newsock);
986 goto error;
987 }
988
989 new_conn = connection_create(newsock, type);
990 if (!new_conn) {
991 lttcomm_destroy_sock(newsock);
992 goto error;
993 }
994
995 /* Enqueue request for the dispatcher thread. */
996 cds_wfcq_enqueue(&relay_conn_queue.head, &relay_conn_queue.tail,
997 &new_conn->qnode);
998
999 /*
1000 * Wake the dispatch queue futex.
1001 * Implicit memory barrier with the
1002 * exchange in cds_wfcq_enqueue.
1003 */
1004 futex_nto1_wake(&relay_conn_queue.futex);
1005 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1006 ERR("socket poll error");
1007 goto error;
1008 } else {
1009 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1010 goto error;
1011 }
1012 }
1013 }
1014
1015 exit:
1016 error:
1017 error_poll_add:
1018 error_testpoint:
1019 lttng_poll_clean(&events);
1020 error_create_poll:
1021 if (data_sock->fd >= 0) {
1022 ret = data_sock->ops->close(data_sock);
1023 if (ret) {
1024 PERROR("close");
1025 }
1026 }
1027 lttcomm_destroy_sock(data_sock);
1028 error_sock_relay:
1029 if (control_sock->fd >= 0) {
1030 ret = control_sock->ops->close(control_sock);
1031 if (ret) {
1032 PERROR("close");
1033 }
1034 }
1035 lttcomm_destroy_sock(control_sock);
1036 error_sock_control:
1037 if (err) {
1038 health_error();
1039 ERR("Health error occurred in %s", __func__);
1040 }
1041 health_unregister(health_relayd);
1042 DBG("Relay listener thread cleanup complete");
1043 lttng_relay_stop_threads();
1044 return NULL;
1045 }
1046
1047 /*
1048 * This thread manages the dispatching of the requests to worker threads
1049 */
1050 static void *relay_thread_dispatcher(void *data)
1051 {
1052 int err = -1;
1053 ssize_t ret;
1054 struct cds_wfcq_node *node;
1055 struct relay_connection *new_conn = NULL;
1056
1057 DBG("[thread] Relay dispatcher started");
1058
1059 health_register(health_relayd, HEALTH_RELAYD_TYPE_DISPATCHER);
1060
1061 if (testpoint(relayd_thread_dispatcher)) {
1062 goto error_testpoint;
1063 }
1064
1065 health_code_update();
1066
1067 for (;;) {
1068 health_code_update();
1069
1070 /* Atomically prepare the queue futex */
1071 futex_nto1_prepare(&relay_conn_queue.futex);
1072
1073 if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
1074 break;
1075 }
1076
1077 do {
1078 health_code_update();
1079
1080 /* Dequeue commands */
1081 node = cds_wfcq_dequeue_blocking(&relay_conn_queue.head,
1082 &relay_conn_queue.tail);
1083 if (node == NULL) {
1084 DBG("Woken up but nothing in the relay command queue");
1085 /* Continue thread execution */
1086 break;
1087 }
1088 new_conn = caa_container_of(node, struct relay_connection, qnode);
1089
1090 DBG("Dispatching request waiting on sock %d", new_conn->sock->fd);
1091
1092 /*
1093 * Inform worker thread of the new request. This
1094 * call is blocking so we can be assured that
1095 * the data will be read at some point in time
1096 * or wait to the end of the world :)
1097 */
1098 ret = lttng_write(relay_conn_pipe[1], &new_conn, sizeof(new_conn));
1099 if (ret < 0) {
1100 PERROR("write connection pipe");
1101 connection_put(new_conn);
1102 goto error;
1103 }
1104 } while (node != NULL);
1105
1106 /* Futex wait on queue. Blocking call on futex() */
1107 health_poll_entry();
1108 futex_nto1_wait(&relay_conn_queue.futex);
1109 health_poll_exit();
1110 }
1111
1112 /* Normal exit, no error */
1113 err = 0;
1114
1115 error:
1116 error_testpoint:
1117 if (err) {
1118 health_error();
1119 ERR("Health error occurred in %s", __func__);
1120 }
1121 health_unregister(health_relayd);
1122 DBG("Dispatch thread dying");
1123 lttng_relay_stop_threads();
1124 return NULL;
1125 }
1126
1127 static bool session_streams_have_index(const struct relay_session *session)
1128 {
1129 return session->minor >= 4 && !session->snapshot;
1130 }
1131
1132 /*
1133 * Handle the RELAYD_CREATE_SESSION command.
1134 *
1135 * On success, send back the session id or else return a negative value.
1136 */
1137 static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr,
1138 struct relay_connection *conn,
1139 const struct lttng_buffer_view *payload)
1140 {
1141 int ret = 0;
1142 ssize_t send_ret;
1143 struct relay_session *session = NULL;
1144 struct lttcomm_relayd_create_session_reply_2_11 reply = {};
1145 char session_name[LTTNG_NAME_MAX] = {};
1146 char hostname[LTTNG_HOST_NAME_MAX] = {};
1147 uint32_t live_timer = 0;
1148 bool snapshot = false;
1149 bool session_name_contains_creation_timestamp = false;
1150 /* Left nil for peers < 2.11. */
1151 char base_path[LTTNG_PATH_MAX] = {};
1152 lttng_uuid sessiond_uuid = {};
1153 LTTNG_OPTIONAL(uint64_t) id_sessiond = {};
1154 LTTNG_OPTIONAL(uint64_t) current_chunk_id = {};
1155 LTTNG_OPTIONAL(time_t) creation_time = {};
1156 struct lttng_dynamic_buffer reply_payload;
1157
1158 lttng_dynamic_buffer_init(&reply_payload);
1159
1160 if (conn->minor < 4) {
1161 /* From 2.1 to 2.3 */
1162 ret = 0;
1163 } else if (conn->minor >= 4 && conn->minor < 11) {
1164 /* From 2.4 to 2.10 */
1165 ret = cmd_create_session_2_4(payload, session_name,
1166 hostname, &live_timer, &snapshot);
1167 } else {
1168 bool has_current_chunk;
1169 uint64_t current_chunk_id_value;
1170 time_t creation_time_value;
1171 uint64_t id_sessiond_value;
1172
1173 /* From 2.11 to ... */
1174 ret = cmd_create_session_2_11(payload, session_name, hostname,
1175 base_path, &live_timer, &snapshot, &id_sessiond_value,
1176 sessiond_uuid, &has_current_chunk,
1177 &current_chunk_id_value, &creation_time_value,
1178 &session_name_contains_creation_timestamp);
1179 if (lttng_uuid_is_nil(sessiond_uuid)) {
1180 /* The nil UUID is reserved for pre-2.11 clients. */
1181 ERR("Illegal nil UUID announced by peer in create session command");
1182 ret = -1;
1183 goto send_reply;
1184 }
1185 LTTNG_OPTIONAL_SET(&id_sessiond, id_sessiond_value);
1186 LTTNG_OPTIONAL_SET(&creation_time, creation_time_value);
1187 if (has_current_chunk) {
1188 LTTNG_OPTIONAL_SET(&current_chunk_id,
1189 current_chunk_id_value);
1190 }
1191 }
1192
1193 if (ret < 0) {
1194 goto send_reply;
1195 }
1196
1197 session = session_create(session_name, hostname, base_path, live_timer,
1198 snapshot, sessiond_uuid,
1199 id_sessiond.is_set ? &id_sessiond.value : NULL,
1200 current_chunk_id.is_set ? &current_chunk_id.value : NULL,
1201 creation_time.is_set ? &creation_time.value : NULL,
1202 conn->major, conn->minor,
1203 session_name_contains_creation_timestamp);
1204 if (!session) {
1205 ret = -1;
1206 goto send_reply;
1207 }
1208 assert(!conn->session);
1209 conn->session = session;
1210 DBG("Created session %" PRIu64, session->id);
1211
1212 reply.generic.session_id = htobe64(session->id);
1213
1214 send_reply:
1215 if (ret < 0) {
1216 reply.generic.ret_code = htobe32(LTTNG_ERR_FATAL);
1217 } else {
1218 reply.generic.ret_code = htobe32(LTTNG_OK);
1219 }
1220
1221 if (conn->minor < 11) {
1222 /* From 2.1 to 2.10 */
1223 ret = lttng_dynamic_buffer_append(&reply_payload,
1224 &reply.generic, sizeof(reply.generic));
1225 if (ret) {
1226 ERR("Failed to append \"create session\" command reply header to payload buffer");
1227 ret = -1;
1228 goto end;
1229 }
1230 } else {
1231 const uint32_t output_path_length =
1232 session ? strlen(session->output_path) + 1 : 0;
1233
1234 reply.output_path_length = htobe32(output_path_length);
1235 ret = lttng_dynamic_buffer_append(
1236 &reply_payload, &reply, sizeof(reply));
1237 if (ret) {
1238 ERR("Failed to append \"create session\" command reply header to payload buffer");
1239 goto end;
1240 }
1241
1242 if (output_path_length) {
1243 ret = lttng_dynamic_buffer_append(&reply_payload,
1244 session->output_path,
1245 output_path_length);
1246 if (ret) {
1247 ERR("Failed to append \"create session\" command reply path to payload buffer");
1248 goto end;
1249 }
1250 }
1251 }
1252
1253 send_ret = conn->sock->ops->sendmsg(conn->sock, reply_payload.data,
1254 reply_payload.size, 0);
1255 if (send_ret < (ssize_t) reply_payload.size) {
1256 ERR("Failed to send \"create session\" command reply of %zu bytes (ret = %zd)",
1257 reply_payload.size, send_ret);
1258 ret = -1;
1259 }
1260 end:
1261 if (ret < 0 && session) {
1262 session_put(session);
1263 }
1264 lttng_dynamic_buffer_reset(&reply_payload);
1265 return ret;
1266 }
1267
1268 /*
1269 * When we have received all the streams and the metadata for a channel,
1270 * we make them visible to the viewer threads.
1271 */
1272 static void publish_connection_local_streams(struct relay_connection *conn)
1273 {
1274 struct relay_stream *stream;
1275 struct relay_session *session = conn->session;
1276
1277 /*
1278 * We publish all streams belonging to a session atomically wrt
1279 * session lock.
1280 */
1281 pthread_mutex_lock(&session->lock);
1282 rcu_read_lock();
1283 cds_list_for_each_entry_rcu(stream, &session->recv_list,
1284 recv_node) {
1285 stream_publish(stream);
1286 }
1287 rcu_read_unlock();
1288
1289 /*
1290 * Inform the viewer that there are new streams in the session.
1291 */
1292 if (session->viewer_attached) {
1293 uatomic_set(&session->new_streams, 1);
1294 }
1295 pthread_mutex_unlock(&session->lock);
1296 }
1297
1298 static int conform_channel_path(char *channel_path)
1299 {
1300 int ret = 0;
1301
1302 if (strstr("../", channel_path)) {
1303 ERR("Refusing channel path as it walks up the path hierarchy: \"%s\"",
1304 channel_path);
1305 ret = -1;
1306 goto end;
1307 }
1308
1309 if (*channel_path == '/') {
1310 const size_t len = strlen(channel_path);
1311
1312 /*
1313 * Channel paths from peers prior to 2.11 are expressed as an
1314 * absolute path that is, in reality, relative to the relay
1315 * daemon's output directory. Remove the leading slash so it
1316 * is correctly interpreted as a relative path later on.
1317 *
1318 * len (and not len - 1) is used to copy the trailing NULL.
1319 */
1320 bcopy(channel_path + 1, channel_path, len);
1321 }
1322 end:
1323 return ret;
1324 }
1325
1326 /*
1327 * relay_add_stream: allocate a new stream for a session
1328 */
1329 static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
1330 struct relay_connection *conn,
1331 const struct lttng_buffer_view *payload)
1332 {
1333 int ret;
1334 ssize_t send_ret;
1335 struct relay_session *session = conn->session;
1336 struct relay_stream *stream = NULL;
1337 struct lttcomm_relayd_status_stream reply;
1338 struct ctf_trace *trace = NULL;
1339 uint64_t stream_handle = -1ULL;
1340 char *path_name = NULL, *channel_name = NULL;
1341 uint64_t tracefile_size = 0, tracefile_count = 0;
1342 LTTNG_OPTIONAL(uint64_t) stream_chunk_id = {};
1343
1344 if (!session || !conn->version_check_done) {
1345 ERR("Trying to add a stream before version check");
1346 ret = -1;
1347 goto end_no_session;
1348 }
1349
1350 if (session->minor == 1) {
1351 /* For 2.1 */
1352 ret = cmd_recv_stream_2_1(payload, &path_name,
1353 &channel_name);
1354 } else if (session->minor > 1 && session->minor < 11) {
1355 /* From 2.2 to 2.10 */
1356 ret = cmd_recv_stream_2_2(payload, &path_name,
1357 &channel_name, &tracefile_size, &tracefile_count);
1358 } else {
1359 /* From 2.11 to ... */
1360 ret = cmd_recv_stream_2_11(payload, &path_name,
1361 &channel_name, &tracefile_size, &tracefile_count,
1362 &stream_chunk_id.value);
1363 stream_chunk_id.is_set = true;
1364 }
1365
1366 if (ret < 0) {
1367 goto send_reply;
1368 }
1369
1370 if (conform_channel_path(path_name)) {
1371 goto send_reply;
1372 }
1373
1374 /*
1375 * Backward compatibility for --group-output-by-session.
1376 * Prior to lttng 2.11, the complete path is passed by the stream.
1377 * Starting at 2.11, lttng-relayd uses chunk. When dealing with producer
1378 * >=2.11 the chunk is responsible for the output path. When dealing
1379 * with producer < 2.11 the chunk output_path is the root output path
1380 * and the stream carries the complete path (path_name).
1381 * To support --group-output-by-session with older producer (<2.11), we
1382 * need to craft the path based on the stream path.
1383 */
1384 if (opt_group_output_by == RELAYD_GROUP_OUTPUT_BY_SESSION) {
1385 if (conn->minor < 4) {
1386 /*
1387 * From 2.1 to 2.3, the session_name is not passed on
1388 * the RELAYD_CREATE_SESSION command. The session name
1389 * is necessary to detect the presence of a base_path
1390 * inside the stream path. Without it we cannot perform
1391 * a valid group-output-by-session transformation.
1392 */
1393 WARN("Unable to perform a --group-by-session transformation for session %" PRIu64
1394 " for stream with path \"%s\" as it is produced by a peer using a protocol older than v2.4",
1395 session->id, path_name);
1396 } else if (conn->minor >= 4 && conn->minor < 11) {
1397 char *group_by_session_path_name;
1398
1399 assert(session->session_name[0] != '\0');
1400
1401 group_by_session_path_name =
1402 backward_compat_group_by_session(
1403 path_name,
1404 session->session_name);
1405 if (!group_by_session_path_name) {
1406 ERR("Failed to apply group by session to stream of session %" PRIu64,
1407 session->id);
1408 goto send_reply;
1409 }
1410
1411 DBG("Transformed session path from \"%s\" to \"%s\" to honor per-session name grouping",
1412 path_name, group_by_session_path_name);
1413
1414 free(path_name);
1415 path_name = group_by_session_path_name;
1416 }
1417 }
1418
1419 trace = ctf_trace_get_by_path_or_create(session, path_name);
1420 if (!trace) {
1421 goto send_reply;
1422 }
1423
1424 /* This stream here has one reference on the trace. */
1425 pthread_mutex_lock(&last_relay_stream_id_lock);
1426 stream_handle = ++last_relay_stream_id;
1427 pthread_mutex_unlock(&last_relay_stream_id_lock);
1428
1429 /* We pass ownership of path_name and channel_name. */
1430 stream = stream_create(trace, stream_handle, path_name,
1431 channel_name, tracefile_size, tracefile_count);
1432 path_name = NULL;
1433 channel_name = NULL;
1434
1435 /*
1436 * Streams are the owners of their trace. Reference to trace is
1437 * kept within stream_create().
1438 */
1439 ctf_trace_put(trace);
1440
1441 send_reply:
1442 memset(&reply, 0, sizeof(reply));
1443 reply.handle = htobe64(stream_handle);
1444 if (!stream) {
1445 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1446 } else {
1447 reply.ret_code = htobe32(LTTNG_OK);
1448 }
1449
1450 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1451 sizeof(struct lttcomm_relayd_status_stream), 0);
1452 if (send_ret < (ssize_t) sizeof(reply)) {
1453 ERR("Failed to send \"add stream\" command reply (ret = %zd)",
1454 send_ret);
1455 ret = -1;
1456 }
1457
1458 end_no_session:
1459 free(path_name);
1460 free(channel_name);
1461 return ret;
1462 }
1463
1464 /*
1465 * relay_close_stream: close a specific stream
1466 */
1467 static int relay_close_stream(const struct lttcomm_relayd_hdr *recv_hdr,
1468 struct relay_connection *conn,
1469 const struct lttng_buffer_view *payload)
1470 {
1471 int ret;
1472 ssize_t send_ret;
1473 struct relay_session *session = conn->session;
1474 struct lttcomm_relayd_close_stream stream_info;
1475 struct lttcomm_relayd_generic_reply reply;
1476 struct relay_stream *stream;
1477
1478 DBG("Close stream received");
1479
1480 if (!session || !conn->version_check_done) {
1481 ERR("Trying to close a stream before version check");
1482 ret = -1;
1483 goto end_no_session;
1484 }
1485
1486 if (payload->size < sizeof(stream_info)) {
1487 ERR("Unexpected payload size in \"relay_close_stream\": expected >= %zu bytes, got %zu bytes",
1488 sizeof(stream_info), payload->size);
1489 ret = -1;
1490 goto end_no_session;
1491 }
1492 memcpy(&stream_info, payload->data, sizeof(stream_info));
1493 stream_info.stream_id = be64toh(stream_info.stream_id);
1494 stream_info.last_net_seq_num = be64toh(stream_info.last_net_seq_num);
1495
1496 stream = stream_get_by_id(stream_info.stream_id);
1497 if (!stream) {
1498 ret = -1;
1499 goto end;
1500 }
1501
1502 /*
1503 * Set last_net_seq_num before the close flag. Required by data
1504 * pending check.
1505 */
1506 pthread_mutex_lock(&stream->lock);
1507 stream->last_net_seq_num = stream_info.last_net_seq_num;
1508 pthread_mutex_unlock(&stream->lock);
1509
1510 /*
1511 * This is one of the conditions which may trigger a stream close
1512 * with the others being:
1513 * 1) A close command is received for a stream
1514 * 2) The control connection owning the stream is closed
1515 * 3) We have received all of the stream's data _after_ a close
1516 * request.
1517 */
1518 try_stream_close(stream);
1519 if (stream->is_metadata) {
1520 struct relay_viewer_stream *vstream;
1521
1522 vstream = viewer_stream_get_by_id(stream->stream_handle);
1523 if (vstream) {
1524 if (vstream->metadata_sent == stream->metadata_received) {
1525 /*
1526 * Since all the metadata has been sent to the
1527 * viewer and that we have a request to close
1528 * its stream, we can safely teardown the
1529 * corresponding metadata viewer stream.
1530 */
1531 viewer_stream_put(vstream);
1532 }
1533 /* Put local reference. */
1534 viewer_stream_put(vstream);
1535 }
1536 }
1537 stream_put(stream);
1538 ret = 0;
1539
1540 end:
1541 memset(&reply, 0, sizeof(reply));
1542 if (ret < 0) {
1543 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1544 } else {
1545 reply.ret_code = htobe32(LTTNG_OK);
1546 }
1547 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1548 sizeof(struct lttcomm_relayd_generic_reply), 0);
1549 if (send_ret < (ssize_t) sizeof(reply)) {
1550 ERR("Failed to send \"close stream\" command reply (ret = %zd)",
1551 send_ret);
1552 ret = -1;
1553 }
1554
1555 end_no_session:
1556 return ret;
1557 }
1558
1559 /*
1560 * relay_reset_metadata: reset a metadata stream
1561 */
1562 static
1563 int relay_reset_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1564 struct relay_connection *conn,
1565 const struct lttng_buffer_view *payload)
1566 {
1567 int ret;
1568 ssize_t send_ret;
1569 struct relay_session *session = conn->session;
1570 struct lttcomm_relayd_reset_metadata stream_info;
1571 struct lttcomm_relayd_generic_reply reply;
1572 struct relay_stream *stream;
1573
1574 DBG("Reset metadata received");
1575
1576 if (!session || !conn->version_check_done) {
1577 ERR("Trying to reset a metadata stream before version check");
1578 ret = -1;
1579 goto end_no_session;
1580 }
1581
1582 if (payload->size < sizeof(stream_info)) {
1583 ERR("Unexpected payload size in \"relay_reset_metadata\": expected >= %zu bytes, got %zu bytes",
1584 sizeof(stream_info), payload->size);
1585 ret = -1;
1586 goto end_no_session;
1587 }
1588 memcpy(&stream_info, payload->data, sizeof(stream_info));
1589 stream_info.stream_id = be64toh(stream_info.stream_id);
1590 stream_info.version = be64toh(stream_info.version);
1591
1592 DBG("Update metadata to version %" PRIu64, stream_info.version);
1593
1594 /* Unsupported for live sessions for now. */
1595 if (session->live_timer != 0) {
1596 ret = -1;
1597 goto end;
1598 }
1599
1600 stream = stream_get_by_id(stream_info.stream_id);
1601 if (!stream) {
1602 ret = -1;
1603 goto end;
1604 }
1605 pthread_mutex_lock(&stream->lock);
1606 if (!stream->is_metadata) {
1607 ret = -1;
1608 goto end_unlock;
1609 }
1610
1611 ret = stream_reset_file(stream);
1612 if (ret < 0) {
1613 ERR("Failed to reset metadata stream %" PRIu64
1614 ": stream_path = %s, channel = %s",
1615 stream->stream_handle, stream->path_name,
1616 stream->channel_name);
1617 goto end_unlock;
1618 }
1619 end_unlock:
1620 pthread_mutex_unlock(&stream->lock);
1621 stream_put(stream);
1622
1623 end:
1624 memset(&reply, 0, sizeof(reply));
1625 if (ret < 0) {
1626 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1627 } else {
1628 reply.ret_code = htobe32(LTTNG_OK);
1629 }
1630 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1631 sizeof(struct lttcomm_relayd_generic_reply), 0);
1632 if (send_ret < (ssize_t) sizeof(reply)) {
1633 ERR("Failed to send \"reset metadata\" command reply (ret = %zd)",
1634 send_ret);
1635 ret = -1;
1636 }
1637
1638 end_no_session:
1639 return ret;
1640 }
1641
1642 /*
1643 * relay_unknown_command: send -1 if received unknown command
1644 */
1645 static void relay_unknown_command(struct relay_connection *conn)
1646 {
1647 struct lttcomm_relayd_generic_reply reply;
1648 ssize_t send_ret;
1649
1650 memset(&reply, 0, sizeof(reply));
1651 reply.ret_code = htobe32(LTTNG_ERR_UNK);
1652 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1653 if (send_ret < sizeof(reply)) {
1654 ERR("Failed to send \"unknown command\" command reply (ret = %zd)", send_ret);
1655 }
1656 }
1657
1658 /*
1659 * relay_start: send an acknowledgment to the client to tell if we are
1660 * ready to receive data. We are ready if a session is established.
1661 */
1662 static int relay_start(const struct lttcomm_relayd_hdr *recv_hdr,
1663 struct relay_connection *conn,
1664 const struct lttng_buffer_view *payload)
1665 {
1666 int ret = 0;
1667 ssize_t send_ret;
1668 struct lttcomm_relayd_generic_reply reply;
1669 struct relay_session *session = conn->session;
1670
1671 if (!session) {
1672 DBG("Trying to start the streaming without a session established");
1673 ret = htobe32(LTTNG_ERR_UNK);
1674 }
1675
1676 memset(&reply, 0, sizeof(reply));
1677 reply.ret_code = htobe32(LTTNG_OK);
1678 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1679 sizeof(reply), 0);
1680 if (send_ret < (ssize_t) sizeof(reply)) {
1681 ERR("Failed to send \"relay_start\" command reply (ret = %zd)",
1682 send_ret);
1683 ret = -1;
1684 }
1685
1686 return ret;
1687 }
1688
1689 /*
1690 * relay_recv_metadata: receive the metadata for the session.
1691 */
1692 static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr,
1693 struct relay_connection *conn,
1694 const struct lttng_buffer_view *payload)
1695 {
1696 int ret = 0;
1697 struct relay_session *session = conn->session;
1698 struct lttcomm_relayd_metadata_payload metadata_payload_header;
1699 struct relay_stream *metadata_stream;
1700 uint64_t metadata_payload_size;
1701 struct lttng_buffer_view packet_view;
1702
1703 if (!session) {
1704 ERR("Metadata sent before version check");
1705 ret = -1;
1706 goto end;
1707 }
1708
1709 if (recv_hdr->data_size < sizeof(struct lttcomm_relayd_metadata_payload)) {
1710 ERR("Incorrect data size");
1711 ret = -1;
1712 goto end;
1713 }
1714 metadata_payload_size = recv_hdr->data_size -
1715 sizeof(struct lttcomm_relayd_metadata_payload);
1716
1717 memcpy(&metadata_payload_header, payload->data,
1718 sizeof(metadata_payload_header));
1719 metadata_payload_header.stream_id = be64toh(
1720 metadata_payload_header.stream_id);
1721 metadata_payload_header.padding_size = be32toh(
1722 metadata_payload_header.padding_size);
1723
1724 metadata_stream = stream_get_by_id(metadata_payload_header.stream_id);
1725 if (!metadata_stream) {
1726 ret = -1;
1727 goto end;
1728 }
1729
1730 packet_view = lttng_buffer_view_from_view(payload,
1731 sizeof(metadata_payload_header), metadata_payload_size);
1732 if (!packet_view.data) {
1733 ERR("Invalid metadata packet length announced by header");
1734 ret = -1;
1735 goto end_put;
1736 }
1737
1738 pthread_mutex_lock(&metadata_stream->lock);
1739 ret = stream_write(metadata_stream, &packet_view,
1740 metadata_payload_header.padding_size);
1741 pthread_mutex_unlock(&metadata_stream->lock);
1742 if (ret){
1743 ret = -1;
1744 goto end_put;
1745 }
1746 end_put:
1747 stream_put(metadata_stream);
1748 end:
1749 return ret;
1750 }
1751
1752 /*
1753 * relay_send_version: send relayd version number
1754 */
1755 static int relay_send_version(const struct lttcomm_relayd_hdr *recv_hdr,
1756 struct relay_connection *conn,
1757 const struct lttng_buffer_view *payload)
1758 {
1759 int ret;
1760 ssize_t send_ret;
1761 struct lttcomm_relayd_version reply, msg;
1762 bool compatible = true;
1763
1764 conn->version_check_done = true;
1765
1766 /* Get version from the other side. */
1767 if (payload->size < sizeof(msg)) {
1768 ERR("Unexpected payload size in \"relay_send_version\": expected >= %zu bytes, got %zu bytes",
1769 sizeof(msg), payload->size);
1770 ret = -1;
1771 goto end;
1772 }
1773
1774 memcpy(&msg, payload->data, sizeof(msg));
1775 msg.major = be32toh(msg.major);
1776 msg.minor = be32toh(msg.minor);
1777
1778 memset(&reply, 0, sizeof(reply));
1779 reply.major = RELAYD_VERSION_COMM_MAJOR;
1780 reply.minor = RELAYD_VERSION_COMM_MINOR;
1781
1782 /* Major versions must be the same */
1783 if (reply.major != msg.major) {
1784 DBG("Incompatible major versions (%u vs %u), deleting session",
1785 reply.major, msg.major);
1786 compatible = false;
1787 }
1788
1789 conn->major = reply.major;
1790 /* We adapt to the lowest compatible version */
1791 if (reply.minor <= msg.minor) {
1792 conn->minor = reply.minor;
1793 } else {
1794 conn->minor = msg.minor;
1795 }
1796
1797 reply.major = htobe32(reply.major);
1798 reply.minor = htobe32(reply.minor);
1799 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
1800 sizeof(reply), 0);
1801 if (send_ret < (ssize_t) sizeof(reply)) {
1802 ERR("Failed to send \"send version\" command reply (ret = %zd)",
1803 send_ret);
1804 ret = -1;
1805 goto end;
1806 } else {
1807 ret = 0;
1808 }
1809
1810 if (!compatible) {
1811 ret = -1;
1812 goto end;
1813 }
1814
1815 DBG("Version check done using protocol %u.%u", conn->major,
1816 conn->minor);
1817
1818 end:
1819 return ret;
1820 }
1821
1822 /*
1823 * Check for data pending for a given stream id from the session daemon.
1824 */
1825 static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
1826 struct relay_connection *conn,
1827 const struct lttng_buffer_view *payload)
1828 {
1829 struct relay_session *session = conn->session;
1830 struct lttcomm_relayd_data_pending msg;
1831 struct lttcomm_relayd_generic_reply reply;
1832 struct relay_stream *stream;
1833 ssize_t send_ret;
1834 int ret;
1835 uint64_t stream_seq;
1836
1837 DBG("Data pending command received");
1838
1839 if (!session || !conn->version_check_done) {
1840 ERR("Trying to check for data before version check");
1841 ret = -1;
1842 goto end_no_session;
1843 }
1844
1845 if (payload->size < sizeof(msg)) {
1846 ERR("Unexpected payload size in \"relay_data_pending\": expected >= %zu bytes, got %zu bytes",
1847 sizeof(msg), payload->size);
1848 ret = -1;
1849 goto end_no_session;
1850 }
1851 memcpy(&msg, payload->data, sizeof(msg));
1852 msg.stream_id = be64toh(msg.stream_id);
1853 msg.last_net_seq_num = be64toh(msg.last_net_seq_num);
1854
1855 stream = stream_get_by_id(msg.stream_id);
1856 if (stream == NULL) {
1857 ret = -1;
1858 goto end;
1859 }
1860
1861 pthread_mutex_lock(&stream->lock);
1862
1863 if (session_streams_have_index(session)) {
1864 /*
1865 * Ensure that both the index and stream data have been
1866 * flushed up to the requested point.
1867 */
1868 stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
1869 } else {
1870 stream_seq = stream->prev_data_seq;
1871 }
1872 DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
1873 ", prev_index_seq %" PRIu64
1874 ", and last_seq %" PRIu64, msg.stream_id,
1875 stream->prev_data_seq, stream->prev_index_seq,
1876 msg.last_net_seq_num);
1877
1878 /* Avoid wrapping issue */
1879 if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
1880 /* Data has in fact been written and is NOT pending */
1881 ret = 0;
1882 } else {
1883 /* Data still being streamed thus pending */
1884 ret = 1;
1885 }
1886
1887 stream->data_pending_check_done = true;
1888 pthread_mutex_unlock(&stream->lock);
1889
1890 stream_put(stream);
1891 end:
1892
1893 memset(&reply, 0, sizeof(reply));
1894 reply.ret_code = htobe32(ret);
1895 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1896 if (send_ret < (ssize_t) sizeof(reply)) {
1897 ERR("Failed to send \"data pending\" command reply (ret = %zd)",
1898 send_ret);
1899 ret = -1;
1900 }
1901
1902 end_no_session:
1903 return ret;
1904 }
1905
1906 /*
1907 * Wait for the control socket to reach a quiescent state.
1908 *
1909 * Note that for now, when receiving this command from the session
1910 * daemon, this means that every subsequent commands or data received on
1911 * the control socket has been handled. So, this is why we simply return
1912 * OK here.
1913 */
1914 static int relay_quiescent_control(const struct lttcomm_relayd_hdr *recv_hdr,
1915 struct relay_connection *conn,
1916 const struct lttng_buffer_view *payload)
1917 {
1918 int ret;
1919 ssize_t send_ret;
1920 struct relay_stream *stream;
1921 struct lttcomm_relayd_quiescent_control msg;
1922 struct lttcomm_relayd_generic_reply reply;
1923
1924 DBG("Checking quiescent state on control socket");
1925
1926 if (!conn->session || !conn->version_check_done) {
1927 ERR("Trying to check for data before version check");
1928 ret = -1;
1929 goto end_no_session;
1930 }
1931
1932 if (payload->size < sizeof(msg)) {
1933 ERR("Unexpected payload size in \"relay_quiescent_control\": expected >= %zu bytes, got %zu bytes",
1934 sizeof(msg), payload->size);
1935 ret = -1;
1936 goto end_no_session;
1937 }
1938 memcpy(&msg, payload->data, sizeof(msg));
1939 msg.stream_id = be64toh(msg.stream_id);
1940
1941 stream = stream_get_by_id(msg.stream_id);
1942 if (!stream) {
1943 goto reply;
1944 }
1945 pthread_mutex_lock(&stream->lock);
1946 stream->data_pending_check_done = true;
1947 pthread_mutex_unlock(&stream->lock);
1948
1949 DBG("Relay quiescent control pending flag set to %" PRIu64, msg.stream_id);
1950 stream_put(stream);
1951 reply:
1952 memset(&reply, 0, sizeof(reply));
1953 reply.ret_code = htobe32(LTTNG_OK);
1954 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
1955 if (send_ret < (ssize_t) sizeof(reply)) {
1956 ERR("Failed to send \"quiescent control\" command reply (ret = %zd)",
1957 send_ret);
1958 ret = -1;
1959 } else {
1960 ret = 0;
1961 }
1962
1963 end_no_session:
1964 return ret;
1965 }
1966
1967 /*
1968 * Initialize a data pending command. This means that a consumer is about
1969 * to ask for data pending for each stream it holds. Simply iterate over
1970 * all streams of a session and set the data_pending_check_done flag.
1971 *
1972 * This command returns to the client a LTTNG_OK code.
1973 */
1974 static int relay_begin_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
1975 struct relay_connection *conn,
1976 const struct lttng_buffer_view *payload)
1977 {
1978 int ret;
1979 ssize_t send_ret;
1980 struct lttng_ht_iter iter;
1981 struct lttcomm_relayd_begin_data_pending msg;
1982 struct lttcomm_relayd_generic_reply reply;
1983 struct relay_stream *stream;
1984
1985 assert(recv_hdr);
1986 assert(conn);
1987
1988 DBG("Init streams for data pending");
1989
1990 if (!conn->session || !conn->version_check_done) {
1991 ERR("Trying to check for data before version check");
1992 ret = -1;
1993 goto end_no_session;
1994 }
1995
1996 if (payload->size < sizeof(msg)) {
1997 ERR("Unexpected payload size in \"relay_begin_data_pending\": expected >= %zu bytes, got %zu bytes",
1998 sizeof(msg), payload->size);
1999 ret = -1;
2000 goto end_no_session;
2001 }
2002 memcpy(&msg, payload->data, sizeof(msg));
2003 msg.session_id = be64toh(msg.session_id);
2004
2005 /*
2006 * Iterate over all streams to set the begin data pending flag.
2007 * For now, the streams are indexed by stream handle so we have
2008 * to iterate over all streams to find the one associated with
2009 * the right session_id.
2010 */
2011 rcu_read_lock();
2012 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2013 node.node) {
2014 if (!stream_get(stream)) {
2015 continue;
2016 }
2017 if (stream->trace->session->id == msg.session_id) {
2018 pthread_mutex_lock(&stream->lock);
2019 stream->data_pending_check_done = false;
2020 pthread_mutex_unlock(&stream->lock);
2021 DBG("Set begin data pending flag to stream %" PRIu64,
2022 stream->stream_handle);
2023 }
2024 stream_put(stream);
2025 }
2026 rcu_read_unlock();
2027
2028 memset(&reply, 0, sizeof(reply));
2029 /* All good, send back reply. */
2030 reply.ret_code = htobe32(LTTNG_OK);
2031
2032 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2033 if (send_ret < (ssize_t) sizeof(reply)) {
2034 ERR("Failed to send \"begin data pending\" command reply (ret = %zd)",
2035 send_ret);
2036 ret = -1;
2037 } else {
2038 ret = 0;
2039 }
2040
2041 end_no_session:
2042 return ret;
2043 }
2044
2045 /*
2046 * End data pending command. This will check, for a given session id, if
2047 * each stream associated with it has its data_pending_check_done flag
2048 * set. If not, this means that the client lost track of the stream but
2049 * the data is still being streamed on our side. In this case, we inform
2050 * the client that data is in flight.
2051 *
2052 * Return to the client if there is data in flight or not with a ret_code.
2053 */
2054 static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
2055 struct relay_connection *conn,
2056 const struct lttng_buffer_view *payload)
2057 {
2058 int ret;
2059 ssize_t send_ret;
2060 struct lttng_ht_iter iter;
2061 struct lttcomm_relayd_end_data_pending msg;
2062 struct lttcomm_relayd_generic_reply reply;
2063 struct relay_stream *stream;
2064 uint32_t is_data_inflight = 0;
2065
2066 DBG("End data pending command");
2067
2068 if (!conn->session || !conn->version_check_done) {
2069 ERR("Trying to check for data before version check");
2070 ret = -1;
2071 goto end_no_session;
2072 }
2073
2074 if (payload->size < sizeof(msg)) {
2075 ERR("Unexpected payload size in \"relay_end_data_pending\": expected >= %zu bytes, got %zu bytes",
2076 sizeof(msg), payload->size);
2077 ret = -1;
2078 goto end_no_session;
2079 }
2080 memcpy(&msg, payload->data, sizeof(msg));
2081 msg.session_id = be64toh(msg.session_id);
2082
2083 /*
2084 * Iterate over all streams to see if the begin data pending
2085 * flag is set.
2086 */
2087 rcu_read_lock();
2088 cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, stream,
2089 node.node) {
2090 if (!stream_get(stream)) {
2091 continue;
2092 }
2093 if (stream->trace->session->id != msg.session_id) {
2094 stream_put(stream);
2095 continue;
2096 }
2097 pthread_mutex_lock(&stream->lock);
2098 if (!stream->data_pending_check_done) {
2099 uint64_t stream_seq;
2100
2101 if (session_streams_have_index(conn->session)) {
2102 /*
2103 * Ensure that both the index and stream data have been
2104 * flushed up to the requested point.
2105 */
2106 stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
2107 } else {
2108 stream_seq = stream->prev_data_seq;
2109 }
2110 if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
2111 is_data_inflight = 1;
2112 DBG("Data is still in flight for stream %" PRIu64,
2113 stream->stream_handle);
2114 pthread_mutex_unlock(&stream->lock);
2115 stream_put(stream);
2116 break;
2117 }
2118 }
2119 pthread_mutex_unlock(&stream->lock);
2120 stream_put(stream);
2121 }
2122 rcu_read_unlock();
2123
2124 memset(&reply, 0, sizeof(reply));
2125 /* All good, send back reply. */
2126 reply.ret_code = htobe32(is_data_inflight);
2127
2128 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2129 if (send_ret < (ssize_t) sizeof(reply)) {
2130 ERR("Failed to send \"end data pending\" command reply (ret = %zd)",
2131 send_ret);
2132 ret = -1;
2133 } else {
2134 ret = 0;
2135 }
2136
2137 end_no_session:
2138 return ret;
2139 }
2140
2141 /*
2142 * Receive an index for a specific stream.
2143 *
2144 * Return 0 on success else a negative value.
2145 */
2146 static int relay_recv_index(const struct lttcomm_relayd_hdr *recv_hdr,
2147 struct relay_connection *conn,
2148 const struct lttng_buffer_view *payload)
2149 {
2150 int ret;
2151 ssize_t send_ret;
2152 struct relay_session *session = conn->session;
2153 struct lttcomm_relayd_index index_info;
2154 struct lttcomm_relayd_generic_reply reply;
2155 struct relay_stream *stream;
2156 size_t msg_len;
2157
2158 assert(conn);
2159
2160 DBG("Relay receiving index");
2161
2162 if (!session || !conn->version_check_done) {
2163 ERR("Trying to close a stream before version check");
2164 ret = -1;
2165 goto end_no_session;
2166 }
2167
2168 msg_len = lttcomm_relayd_index_len(
2169 lttng_to_index_major(conn->major, conn->minor),
2170 lttng_to_index_minor(conn->major, conn->minor));
2171 if (payload->size < msg_len) {
2172 ERR("Unexpected payload size in \"relay_recv_index\": expected >= %zu bytes, got %zu bytes",
2173 msg_len, payload->size);
2174 ret = -1;
2175 goto end_no_session;
2176 }
2177 memcpy(&index_info, payload->data, msg_len);
2178 index_info.relay_stream_id = be64toh(index_info.relay_stream_id);
2179 index_info.net_seq_num = be64toh(index_info.net_seq_num);
2180 index_info.packet_size = be64toh(index_info.packet_size);
2181 index_info.content_size = be64toh(index_info.content_size);
2182 index_info.timestamp_begin = be64toh(index_info.timestamp_begin);
2183 index_info.timestamp_end = be64toh(index_info.timestamp_end);
2184 index_info.events_discarded = be64toh(index_info.events_discarded);
2185 index_info.stream_id = be64toh(index_info.stream_id);
2186
2187 if (conn->minor >= 8) {
2188 index_info.stream_instance_id =
2189 be64toh(index_info.stream_instance_id);
2190 index_info.packet_seq_num = be64toh(index_info.packet_seq_num);
2191 }
2192
2193 stream = stream_get_by_id(index_info.relay_stream_id);
2194 if (!stream) {
2195 ERR("stream_get_by_id not found");
2196 ret = -1;
2197 goto end;
2198 }
2199
2200 pthread_mutex_lock(&stream->lock);
2201 ret = stream_add_index(stream, &index_info);
2202 pthread_mutex_unlock(&stream->lock);
2203 if (ret) {
2204 goto end_stream_put;
2205 }
2206
2207 end_stream_put:
2208 stream_put(stream);
2209 end:
2210 memset(&reply, 0, sizeof(reply));
2211 if (ret < 0) {
2212 reply.ret_code = htobe32(LTTNG_ERR_UNK);
2213 } else {
2214 reply.ret_code = htobe32(LTTNG_OK);
2215 }
2216 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2217 if (send_ret < (ssize_t) sizeof(reply)) {
2218 ERR("Failed to send \"recv index\" command reply (ret = %zd)", send_ret);
2219 ret = -1;
2220 }
2221
2222 end_no_session:
2223 return ret;
2224 }
2225
2226 /*
2227 * Receive the streams_sent message.
2228 *
2229 * Return 0 on success else a negative value.
2230 */
2231 static int relay_streams_sent(const struct lttcomm_relayd_hdr *recv_hdr,
2232 struct relay_connection *conn,
2233 const struct lttng_buffer_view *payload)
2234 {
2235 int ret;
2236 ssize_t send_ret;
2237 struct lttcomm_relayd_generic_reply reply;
2238
2239 assert(conn);
2240
2241 DBG("Relay receiving streams_sent");
2242
2243 if (!conn->session || !conn->version_check_done) {
2244 ERR("Trying to close a stream before version check");
2245 ret = -1;
2246 goto end_no_session;
2247 }
2248
2249 /*
2250 * Publish every pending stream in the connection recv list which are
2251 * now ready to be used by the viewer.
2252 */
2253 publish_connection_local_streams(conn);
2254
2255 memset(&reply, 0, sizeof(reply));
2256 reply.ret_code = htobe32(LTTNG_OK);
2257 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
2258 if (send_ret < (ssize_t) sizeof(reply)) {
2259 ERR("Failed to send \"streams sent\" command reply (ret = %zd)",
2260 send_ret);
2261 ret = -1;
2262 } else {
2263 /* Success. */
2264 ret = 0;
2265 }
2266
2267 end_no_session:
2268 return ret;
2269 }
2270
2271 /*
2272 * relay_rotate_session_stream: rotate a stream to a new tracefile for the
2273 * session rotation feature (not the tracefile rotation feature).
2274 */
2275 static int relay_rotate_session_streams(
2276 const struct lttcomm_relayd_hdr *recv_hdr,
2277 struct relay_connection *conn,
2278 const struct lttng_buffer_view *payload)
2279 {
2280 int ret = 0;
2281 uint32_t i;
2282 ssize_t send_ret;
2283 enum lttng_error_code reply_code = LTTNG_ERR_UNK;
2284 struct relay_session *session = conn->session;
2285 struct lttcomm_relayd_rotate_streams rotate_streams;
2286 struct lttcomm_relayd_generic_reply reply = {};
2287 struct relay_stream *stream = NULL;
2288 const size_t header_len = sizeof(struct lttcomm_relayd_rotate_streams);
2289 struct lttng_trace_chunk *next_trace_chunk = NULL;
2290 struct lttng_buffer_view stream_positions;
2291 char chunk_id_buf[MAX_INT_DEC_LEN(uint64_t)];
2292 const char *chunk_id_str = "none";
2293
2294 if (!session || !conn->version_check_done) {
2295 ERR("Trying to rotate a stream before version check");
2296 ret = -1;
2297 goto end_no_reply;
2298 }
2299
2300 if (session->major == 2 && session->minor < 11) {
2301 ERR("Unsupported feature before 2.11");
2302 ret = -1;
2303 goto end_no_reply;
2304 }
2305
2306 if (payload->size < header_len) {
2307 ERR("Unexpected payload size in \"relay_rotate_session_stream\": expected >= %zu bytes, got %zu bytes",
2308 header_len, payload->size);
2309 ret = -1;
2310 goto end_no_reply;
2311 }
2312
2313 memcpy(&rotate_streams, payload->data, header_len);
2314
2315 /* Convert header to host endianness. */
2316 rotate_streams = (typeof(rotate_streams)) {
2317 .stream_count = be32toh(rotate_streams.stream_count),
2318 .new_chunk_id = (typeof(rotate_streams.new_chunk_id)) {
2319 .is_set = !!rotate_streams.new_chunk_id.is_set,
2320 .value = be64toh(rotate_streams.new_chunk_id.value),
2321 }
2322 };
2323
2324 if (rotate_streams.new_chunk_id.is_set) {
2325 /*
2326 * Retrieve the trace chunk the stream must transition to. As
2327 * per the protocol, this chunk should have been created
2328 * before this command is received.
2329 */
2330 next_trace_chunk = sessiond_trace_chunk_registry_get_chunk(
2331 sessiond_trace_chunk_registry,
2332 session->sessiond_uuid, session->id,
2333 rotate_streams.new_chunk_id.value);
2334 if (!next_trace_chunk) {
2335 char uuid_str[UUID_STR_LEN];
2336
2337 lttng_uuid_to_str(session->sessiond_uuid, uuid_str);
2338 ERR("Unknown next trace chunk in ROTATE_STREAMS command: sessiond_uuid = {%s}, session_id = %" PRIu64
2339 ", trace_chunk_id = %" PRIu64,
2340 uuid_str, session->id,
2341 rotate_streams.new_chunk_id.value);
2342 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2343 ret = -1;
2344 goto end;
2345 }
2346
2347 ret = snprintf(chunk_id_buf, sizeof(chunk_id_buf), "%" PRIu64,
2348 rotate_streams.new_chunk_id.value);
2349 if (ret < 0 || ret >= sizeof(chunk_id_buf)) {
2350 chunk_id_str = "formatting error";
2351 } else {
2352 chunk_id_str = chunk_id_buf;
2353 }
2354 session->has_rotated = true;
2355 }
2356
2357 DBG("Rotate %" PRIu32 " streams of session \"%s\" to chunk \"%s\"",
2358 rotate_streams.stream_count, session->session_name,
2359 chunk_id_str);
2360
2361 stream_positions = lttng_buffer_view_from_view(payload,
2362 sizeof(rotate_streams), -1);
2363 if (!stream_positions.data ||
2364 stream_positions.size <
2365 (rotate_streams.stream_count *
2366 sizeof(struct lttcomm_relayd_stream_rotation_position))) {
2367 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2368 ret = -1;
2369 goto end;
2370 }
2371
2372 for (i = 0; i < rotate_streams.stream_count; i++) {
2373 struct lttcomm_relayd_stream_rotation_position *position_comm =
2374 &((typeof(position_comm)) stream_positions.data)[i];
2375 const struct lttcomm_relayd_stream_rotation_position pos = {
2376 .stream_id = be64toh(position_comm->stream_id),
2377 .rotate_at_seq_num = be64toh(
2378 position_comm->rotate_at_seq_num),
2379 };
2380
2381 stream = stream_get_by_id(pos.stream_id);
2382 if (!stream) {
2383 reply_code = LTTNG_ERR_INVALID;
2384 ret = -1;
2385 goto end;
2386 }
2387
2388 pthread_mutex_lock(&stream->lock);
2389 ret = stream_set_pending_rotation(stream, next_trace_chunk,
2390 pos.rotate_at_seq_num);
2391 pthread_mutex_unlock(&stream->lock);
2392 if (ret) {
2393 reply_code = LTTNG_ERR_FILE_CREATION_ERROR;
2394 goto end;
2395 }
2396
2397 stream_put(stream);
2398 stream = NULL;
2399 }
2400
2401 reply_code = LTTNG_OK;
2402 ret = 0;
2403 end:
2404 if (stream) {
2405 stream_put(stream);
2406 }
2407
2408 reply.ret_code = htobe32((uint32_t) reply_code);
2409 send_ret = conn->sock->ops->sendmsg(conn->sock, &reply,
2410 sizeof(struct lttcomm_relayd_generic_reply), 0);
2411 if (send_ret < (ssize_t) sizeof(reply)) {
2412 ERR("Failed to send \"rotate session stream\" command reply (ret = %zd)",
2413 send_ret);
2414 ret = -1;
2415 }
2416 end_no_reply:
2417 lttng_trace_chunk_put(next_trace_chunk);
2418 return ret;
2419 }
2420
2421
2422
2423 /*
2424 * relay_create_trace_chunk: create a new trace chunk
2425 */
2426 static int relay_create_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr,
2427 struct relay_connection *conn,
2428 const struct lttng_buffer_view *payload)
2429 {
2430 int ret = 0;
2431 ssize_t send_ret;
2432 struct relay_session *session = conn->session;
2433 struct lttcomm_relayd_create_trace_chunk *msg;
2434 struct lttcomm_relayd_generic_reply reply = {};
2435 struct lttng_buffer_view header_view;
2436 struct lttng_buffer_view chunk_name_view;
2437 struct lttng_trace_chunk *chunk = NULL, *published_chunk = NULL;
2438 enum lttng_error_code reply_code = LTTNG_OK;
2439 enum lttng_trace_chunk_status chunk_status;
2440 struct lttng_directory_handle session_output;
2441
2442 if (!session || !conn->version_check_done) {
2443 ERR("Trying to create a trace chunk before version check");
2444 ret = -1;
2445 goto end_no_reply;
2446 }
2447
2448 if (session->major == 2 && session->minor < 11) {
2449 ERR("Chunk creation command is unsupported before 2.11");
2450 ret = -1;
2451 goto end_no_reply;
2452 }
2453
2454 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2455 if (!header_view.data) {
2456 ERR("Failed to receive payload of chunk creation command");
2457 ret = -1;
2458 goto end_no_reply;
2459 }
2460
2461 /* Convert to host endianness. */
2462 msg = (typeof(msg)) header_view.data;
2463 msg->chunk_id = be64toh(msg->chunk_id);
2464 msg->creation_timestamp = be64toh(msg->creation_timestamp);
2465 msg->override_name_length = be32toh(msg->override_name_length);
2466
2467 chunk = lttng_trace_chunk_create(
2468 msg->chunk_id, msg->creation_timestamp);
2469 if (!chunk) {
2470 ERR("Failed to create trace chunk in trace chunk creation command");
2471 ret = -1;
2472 reply_code = LTTNG_ERR_NOMEM;
2473 goto end;
2474 }
2475
2476 if (msg->override_name_length) {
2477 const char *name;
2478
2479 chunk_name_view = lttng_buffer_view_from_view(payload,
2480 sizeof(*msg),
2481 msg->override_name_length);
2482 name = chunk_name_view.data;
2483 if (!name || name[msg->override_name_length - 1]) {
2484 ERR("Failed to receive payload of chunk creation command");
2485 ret = -1;
2486 reply_code = LTTNG_ERR_INVALID;
2487 goto end;
2488 }
2489
2490 chunk_status = lttng_trace_chunk_override_name(
2491 chunk, chunk_name_view.data);
2492 switch (chunk_status) {
2493 case LTTNG_TRACE_CHUNK_STATUS_OK:
2494 break;
2495 case LTTNG_TRACE_CHUNK_STATUS_INVALID_ARGUMENT:
2496 ERR("Failed to set the name of new trace chunk in trace chunk creation command (invalid name)");
2497 reply_code = LTTNG_ERR_INVALID;
2498 ret = -1;
2499 goto end;
2500 default:
2501 ERR("Failed to set the name of new trace chunk in trace chunk creation command (unknown error)");
2502 reply_code = LTTNG_ERR_UNK;
2503 ret = -1;
2504 goto end;
2505 }
2506 }
2507
2508 chunk_status = lttng_trace_chunk_set_credentials_current_user(chunk);
2509 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2510 reply_code = LTTNG_ERR_UNK;
2511 ret = -1;
2512 goto end;
2513 }
2514
2515 ret = session_init_output_directory_handle(
2516 conn->session, &session_output);
2517 if (ret) {
2518 reply_code = LTTNG_ERR_CREATE_DIR_FAIL;
2519 goto end;
2520 }
2521 chunk_status = lttng_trace_chunk_set_as_owner(chunk, &session_output);
2522 lttng_directory_handle_fini(&session_output);
2523 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2524 reply_code = LTTNG_ERR_UNK;
2525 ret = -1;
2526 goto end;
2527 }
2528
2529 published_chunk = sessiond_trace_chunk_registry_publish_chunk(
2530 sessiond_trace_chunk_registry,
2531 conn->session->sessiond_uuid,
2532 conn->session->id,
2533 chunk);
2534 if (!published_chunk) {
2535 char uuid_str[UUID_STR_LEN];
2536
2537 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2538 ERR("Failed to publish chunk: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2539 uuid_str,
2540 conn->session->id,
2541 msg->chunk_id);
2542 ret = -1;
2543 reply_code = LTTNG_ERR_NOMEM;
2544 goto end;
2545 }
2546
2547 pthread_mutex_lock(&conn->session->lock);
2548 if (conn->session->pending_closure_trace_chunk) {
2549 /*
2550 * Invalid; this means a second create_trace_chunk command was
2551 * received before a close_trace_chunk.
2552 */
2553 ERR("Invalid trace chunk close command received; a trace chunk is already waiting for a trace chunk close command");
2554 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2555 ret = -1;
2556 goto end_unlock_session;
2557 }
2558 conn->session->pending_closure_trace_chunk =
2559 conn->session->current_trace_chunk;
2560 conn->session->current_trace_chunk = published_chunk;
2561 published_chunk = NULL;
2562 end_unlock_session:
2563 pthread_mutex_unlock(&conn->session->lock);
2564 end:
2565 reply.ret_code = htobe32((uint32_t) reply_code);
2566 send_ret = conn->sock->ops->sendmsg(conn->sock,
2567 &reply,
2568 sizeof(struct lttcomm_relayd_generic_reply),
2569 0);
2570 if (send_ret < (ssize_t) sizeof(reply)) {
2571 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
2572 send_ret);
2573 ret = -1;
2574 }
2575 end_no_reply:
2576 lttng_trace_chunk_put(chunk);
2577 lttng_trace_chunk_put(published_chunk);
2578 return ret;
2579 }
2580
2581 /*
2582 * relay_close_trace_chunk: close a trace chunk
2583 */
2584 static int relay_close_trace_chunk(const struct lttcomm_relayd_hdr *recv_hdr,
2585 struct relay_connection *conn,
2586 const struct lttng_buffer_view *payload)
2587 {
2588 int ret = 0, buf_ret;
2589 ssize_t send_ret;
2590 struct relay_session *session = conn->session;
2591 struct lttcomm_relayd_close_trace_chunk *msg;
2592 struct lttcomm_relayd_close_trace_chunk_reply reply = {};
2593 struct lttng_buffer_view header_view;
2594 struct lttng_trace_chunk *chunk = NULL;
2595 enum lttng_error_code reply_code = LTTNG_OK;
2596 enum lttng_trace_chunk_status chunk_status;
2597 uint64_t chunk_id;
2598 LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
2599 time_t close_timestamp;
2600 char closed_trace_chunk_path[LTTNG_PATH_MAX];
2601 size_t path_length = 0;
2602 const char *chunk_name = NULL;
2603 struct lttng_dynamic_buffer reply_payload;
2604
2605 lttng_dynamic_buffer_init(&reply_payload);
2606
2607 if (!session || !conn->version_check_done) {
2608 ERR("Trying to close a trace chunk before version check");
2609 ret = -1;
2610 goto end_no_reply;
2611 }
2612
2613 if (session->major == 2 && session->minor < 11) {
2614 ERR("Chunk close command is unsupported before 2.11");
2615 ret = -1;
2616 goto end_no_reply;
2617 }
2618
2619 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2620 if (!header_view.data) {
2621 ERR("Failed to receive payload of chunk close command");
2622 ret = -1;
2623 goto end_no_reply;
2624 }
2625
2626 /* Convert to host endianness. */
2627 msg = (typeof(msg)) header_view.data;
2628 chunk_id = be64toh(msg->chunk_id);
2629 close_timestamp = (time_t) be64toh(msg->close_timestamp);
2630 close_command = (typeof(close_command)){
2631 .value = be32toh(msg->close_command.value),
2632 .is_set = msg->close_command.is_set,
2633 };
2634
2635 chunk = sessiond_trace_chunk_registry_get_chunk(
2636 sessiond_trace_chunk_registry,
2637 conn->session->sessiond_uuid,
2638 conn->session->id,
2639 chunk_id);
2640 if (!chunk) {
2641 char uuid_str[UUID_STR_LEN];
2642
2643 lttng_uuid_to_str(conn->session->sessiond_uuid, uuid_str);
2644 ERR("Failed to find chunk to close: sessiond_uuid = %s, session_id = %" PRIu64 ", chunk_id = %" PRIu64,
2645 uuid_str,
2646 conn->session->id,
2647 msg->chunk_id);
2648 ret = -1;
2649 reply_code = LTTNG_ERR_NOMEM;
2650 goto end;
2651 }
2652
2653 pthread_mutex_lock(&session->lock);
2654 if (session->pending_closure_trace_chunk &&
2655 session->pending_closure_trace_chunk != chunk) {
2656 ERR("Trace chunk close command for session \"%s\" does not target the trace chunk pending closure",
2657 session->session_name);
2658 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2659 ret = -1;
2660 goto end_unlock_session;
2661 }
2662
2663 chunk_status = lttng_trace_chunk_set_close_timestamp(
2664 chunk, close_timestamp);
2665 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2666 ERR("Failed to set trace chunk close timestamp");
2667 ret = -1;
2668 reply_code = LTTNG_ERR_UNK;
2669 goto end_unlock_session;
2670 }
2671
2672 if (close_command.is_set) {
2673 chunk_status = lttng_trace_chunk_set_close_command(
2674 chunk, close_command.value);
2675 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2676 ret = -1;
2677 reply_code = LTTNG_ERR_INVALID;
2678 goto end_unlock_session;
2679 }
2680 }
2681 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name, NULL);
2682 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2683 ERR("Failed to get chunk name");
2684 ret = -1;
2685 reply_code = LTTNG_ERR_UNK;
2686 goto end_unlock_session;
2687 }
2688 if (!session->has_rotated && !session->snapshot) {
2689 ret = lttng_strncpy(closed_trace_chunk_path,
2690 session->output_path,
2691 sizeof(closed_trace_chunk_path));
2692 if (ret) {
2693 ERR("Failed to send trace chunk path: path length of %zu bytes exceeds the maximal allowed length of %zu bytes",
2694 strlen(session->output_path),
2695 sizeof(closed_trace_chunk_path));
2696 reply_code = LTTNG_ERR_NOMEM;
2697 ret = -1;
2698 goto end_unlock_session;
2699 }
2700 } else {
2701 if (session->snapshot) {
2702 ret = snprintf(closed_trace_chunk_path,
2703 sizeof(closed_trace_chunk_path),
2704 "%s/%s", session->output_path,
2705 chunk_name);
2706 } else {
2707 ret = snprintf(closed_trace_chunk_path,
2708 sizeof(closed_trace_chunk_path),
2709 "%s/" DEFAULT_ARCHIVED_TRACE_CHUNKS_DIRECTORY
2710 "/%s",
2711 session->output_path, chunk_name);
2712 }
2713 if (ret < 0 || ret == sizeof(closed_trace_chunk_path)) {
2714 ERR("Failed to format closed trace chunk resulting path");
2715 reply_code = ret < 0 ? LTTNG_ERR_UNK : LTTNG_ERR_NOMEM;
2716 ret = -1;
2717 goto end_unlock_session;
2718 }
2719 }
2720 DBG("Reply chunk path on close: %s", closed_trace_chunk_path);
2721 path_length = strlen(closed_trace_chunk_path) + 1;
2722 if (path_length > UINT32_MAX) {
2723 ERR("Closed trace chunk path exceeds the maximal length allowed by the protocol");
2724 ret = -1;
2725 reply_code = LTTNG_ERR_INVALID_PROTOCOL;
2726 goto end_unlock_session;
2727 }
2728
2729 if (session->current_trace_chunk == chunk) {
2730 /*
2731 * After a trace chunk close command, no new streams
2732 * referencing the chunk may be created. Hence, on the
2733 * event that no new trace chunk have been created for
2734 * the session, the reference to the current trace chunk
2735 * is released in order to allow it to be reclaimed when
2736 * the last stream releases its reference to it.
2737 */
2738 lttng_trace_chunk_put(session->current_trace_chunk);
2739 session->current_trace_chunk = NULL;
2740 }
2741 lttng_trace_chunk_put(session->pending_closure_trace_chunk);
2742 session->pending_closure_trace_chunk = NULL;
2743 end_unlock_session:
2744 pthread_mutex_unlock(&session->lock);
2745
2746 end:
2747 reply.generic.ret_code = htobe32((uint32_t) reply_code);
2748 reply.path_length = htobe32((uint32_t) path_length);
2749 buf_ret = lttng_dynamic_buffer_append(
2750 &reply_payload, &reply, sizeof(reply));
2751 if (buf_ret) {
2752 ERR("Failed to append \"close trace chunk\" command reply header to payload buffer");
2753 goto end_no_reply;
2754 }
2755
2756 if (reply_code == LTTNG_OK) {
2757 buf_ret = lttng_dynamic_buffer_append(&reply_payload,
2758 closed_trace_chunk_path, path_length);
2759 if (buf_ret) {
2760 ERR("Failed to append \"close trace chunk\" command reply path to payload buffer");
2761 goto end_no_reply;
2762 }
2763 }
2764
2765 send_ret = conn->sock->ops->sendmsg(conn->sock,
2766 reply_payload.data,
2767 reply_payload.size,
2768 0);
2769 if (send_ret < reply_payload.size) {
2770 ERR("Failed to send \"close trace chunk\" command reply of %zu bytes (ret = %zd)",
2771 reply_payload.size, send_ret);
2772 ret = -1;
2773 goto end_no_reply;
2774 }
2775 end_no_reply:
2776 lttng_trace_chunk_put(chunk);
2777 lttng_dynamic_buffer_reset(&reply_payload);
2778 return ret;
2779 }
2780
2781 /*
2782 * relay_trace_chunk_exists: check if a trace chunk exists
2783 */
2784 static int relay_trace_chunk_exists(const struct lttcomm_relayd_hdr *recv_hdr,
2785 struct relay_connection *conn,
2786 const struct lttng_buffer_view *payload)
2787 {
2788 int ret = 0;
2789 ssize_t send_ret;
2790 struct relay_session *session = conn->session;
2791 struct lttcomm_relayd_trace_chunk_exists *msg;
2792 struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
2793 struct lttng_buffer_view header_view;
2794 uint64_t chunk_id;
2795 bool chunk_exists;
2796
2797 if (!session || !conn->version_check_done) {
2798 ERR("Trying to close a trace chunk before version check");
2799 ret = -1;
2800 goto end_no_reply;
2801 }
2802
2803 if (session->major == 2 && session->minor < 11) {
2804 ERR("Chunk close command is unsupported before 2.11");
2805 ret = -1;
2806 goto end_no_reply;
2807 }
2808
2809 header_view = lttng_buffer_view_from_view(payload, 0, sizeof(*msg));
2810 if (!header_view.data) {
2811 ERR("Failed to receive payload of chunk close command");
2812 ret = -1;
2813 goto end_no_reply;
2814 }
2815
2816 /* Convert to host endianness. */
2817 msg = (typeof(msg)) header_view.data;
2818 chunk_id = be64toh(msg->chunk_id);
2819
2820 ret = sessiond_trace_chunk_registry_chunk_exists(
2821 sessiond_trace_chunk_registry,
2822 conn->session->sessiond_uuid,
2823 conn->session->id,
2824 chunk_id, &chunk_exists);
2825 /*
2826 * If ret is not 0, send the reply and report the error to the caller.
2827 * It is a protocol (or internal) error and the session/connection
2828 * should be torn down.
2829 */
2830 reply = (typeof(reply)){
2831 .generic.ret_code = htobe32((uint32_t)
2832 (ret == 0 ? LTTNG_OK : LTTNG_ERR_INVALID_PROTOCOL)),
2833 .trace_chunk_exists = ret == 0 ? chunk_exists : 0,
2834 };
2835 send_ret = conn->sock->ops->sendmsg(
2836 conn->sock, &reply, sizeof(reply), 0);
2837 if (send_ret < (ssize_t) sizeof(reply)) {
2838 ERR("Failed to send \"create trace chunk\" command reply (ret = %zd)",
2839 send_ret);
2840 ret = -1;
2841 }
2842 end_no_reply:
2843 return ret;
2844 }
2845
2846 #define DBG_CMD(cmd_name, conn) \
2847 DBG3("Processing \"%s\" command for socket %i", cmd_name, conn->sock->fd);
2848
2849 static int relay_process_control_command(struct relay_connection *conn,
2850 const struct lttcomm_relayd_hdr *header,
2851 const struct lttng_buffer_view *payload)
2852 {
2853 int ret = 0;
2854
2855 switch (header->cmd) {
2856 case RELAYD_CREATE_SESSION:
2857 DBG_CMD("RELAYD_CREATE_SESSION", conn);
2858 ret = relay_create_session(header, conn, payload);
2859 break;
2860 case RELAYD_ADD_STREAM:
2861 DBG_CMD("RELAYD_ADD_STREAM", conn);
2862 ret = relay_add_stream(header, conn, payload);
2863 break;
2864 case RELAYD_START_DATA:
2865 DBG_CMD("RELAYD_START_DATA", conn);
2866 ret = relay_start(header, conn, payload);
2867 break;
2868 case RELAYD_SEND_METADATA:
2869 DBG_CMD("RELAYD_SEND_METADATA", conn);
2870 ret = relay_recv_metadata(header, conn, payload);
2871 break;
2872 case RELAYD_VERSION:
2873 DBG_CMD("RELAYD_VERSION", conn);
2874 ret = relay_send_version(header, conn, payload);
2875 break;
2876 case RELAYD_CLOSE_STREAM:
2877 DBG_CMD("RELAYD_CLOSE_STREAM", conn);
2878 ret = relay_close_stream(header, conn, payload);
2879 break;
2880 case RELAYD_DATA_PENDING:
2881 DBG_CMD("RELAYD_DATA_PENDING", conn);
2882 ret = relay_data_pending(header, conn, payload);
2883 break;
2884 case RELAYD_QUIESCENT_CONTROL:
2885 DBG_CMD("RELAYD_QUIESCENT_CONTROL", conn);
2886 ret = relay_quiescent_control(header, conn, payload);
2887 break;
2888 case RELAYD_BEGIN_DATA_PENDING:
2889 DBG_CMD("RELAYD_BEGIN_DATA_PENDING", conn);
2890 ret = relay_begin_data_pending(header, conn, payload);
2891 break;
2892 case RELAYD_END_DATA_PENDING:
2893 DBG_CMD("RELAYD_END_DATA_PENDING", conn);
2894 ret = relay_end_data_pending(header, conn, payload);
2895 break;
2896 case RELAYD_SEND_INDEX:
2897 DBG_CMD("RELAYD_SEND_INDEX", conn);
2898 ret = relay_recv_index(header, conn, payload);
2899 break;
2900 case RELAYD_STREAMS_SENT:
2901 DBG_CMD("RELAYD_STREAMS_SENT", conn);
2902 ret = relay_streams_sent(header, conn, payload);
2903 break;
2904 case RELAYD_RESET_METADATA:
2905 DBG_CMD("RELAYD_RESET_METADATA", conn);
2906 ret = relay_reset_metadata(header, conn, payload);
2907 break;
2908 case RELAYD_ROTATE_STREAMS:
2909 DBG_CMD("RELAYD_ROTATE_STREAMS", conn);
2910 ret = relay_rotate_session_streams(header, conn, payload);
2911 break;
2912 case RELAYD_CREATE_TRACE_CHUNK:
2913 DBG_CMD("RELAYD_CREATE_TRACE_CHUNK", conn);
2914 ret = relay_create_trace_chunk(header, conn, payload);
2915 break;
2916 case RELAYD_CLOSE_TRACE_CHUNK:
2917 DBG_CMD("RELAYD_CLOSE_TRACE_CHUNK", conn);
2918 ret = relay_close_trace_chunk(header, conn, payload);
2919 break;
2920 case RELAYD_TRACE_CHUNK_EXISTS:
2921 DBG_CMD("RELAYD_TRACE_CHUNK_EXISTS", conn);
2922 ret = relay_trace_chunk_exists(header, conn, payload);
2923 break;
2924 case RELAYD_UPDATE_SYNC_INFO:
2925 default:
2926 ERR("Received unknown command (%u)", header->cmd);
2927 relay_unknown_command(conn);
2928 ret = -1;
2929 goto end;
2930 }
2931
2932 end:
2933 return ret;
2934 }
2935
2936 static enum relay_connection_status relay_process_control_receive_payload(
2937 struct relay_connection *conn)
2938 {
2939 int ret = 0;
2940 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
2941 struct lttng_dynamic_buffer *reception_buffer =
2942 &conn->protocol.ctrl.reception_buffer;
2943 struct ctrl_connection_state_receive_payload *state =
2944 &conn->protocol.ctrl.state.receive_payload;
2945 struct lttng_buffer_view payload_view;
2946
2947 if (state->left_to_receive == 0) {
2948 /* Short-circuit for payload-less commands. */
2949 goto reception_complete;
2950 }
2951
2952 ret = conn->sock->ops->recvmsg(conn->sock,
2953 reception_buffer->data + state->received,
2954 state->left_to_receive, MSG_DONTWAIT);
2955 if (ret < 0) {
2956 if (errno != EAGAIN && errno != EWOULDBLOCK) {
2957 PERROR("Unable to receive command payload on sock %d",
2958 conn->sock->fd);
2959 status = RELAY_CONNECTION_STATUS_ERROR;
2960 }
2961 goto end;
2962 } else if (ret == 0) {
2963 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
2964 status = RELAY_CONNECTION_STATUS_CLOSED;
2965 goto end;
2966 }
2967
2968 assert(ret > 0);
2969 assert(ret <= state->left_to_receive);
2970
2971 state->left_to_receive -= ret;
2972 state->received += ret;
2973
2974 if (state->left_to_receive > 0) {
2975 /*
2976 * Can't transition to the protocol's next state, wait to
2977 * receive the rest of the header.
2978 */
2979 DBG3("Partial reception of control connection protocol payload (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
2980 state->received, state->left_to_receive,
2981 conn->sock->fd);
2982 goto end;
2983 }
2984
2985 reception_complete:
2986 DBG("Done receiving control command payload: fd = %i, payload size = %" PRIu64 " bytes",
2987 conn->sock->fd, state->received);
2988 /*
2989 * The payload required to process the command has been received.
2990 * A view to the reception buffer is forwarded to the various
2991 * commands and the state of the control is reset on success.
2992 *
2993 * Commands are responsible for sending their reply to the peer.
2994 */
2995 payload_view = lttng_buffer_view_from_dynamic_buffer(reception_buffer,
2996 0, -1);
2997 ret = relay_process_control_command(conn,
2998 &state->header, &payload_view);
2999 if (ret < 0) {
3000 status = RELAY_CONNECTION_STATUS_ERROR;
3001 goto end;
3002 }
3003
3004 ret = connection_reset_protocol_state(conn);
3005 if (ret) {
3006 status = RELAY_CONNECTION_STATUS_ERROR;
3007 }
3008 end:
3009 return status;
3010 }
3011
3012 static enum relay_connection_status relay_process_control_receive_header(
3013 struct relay_connection *conn)
3014 {
3015 int ret = 0;
3016 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3017 struct lttcomm_relayd_hdr header;
3018 struct lttng_dynamic_buffer *reception_buffer =
3019 &conn->protocol.ctrl.reception_buffer;
3020 struct ctrl_connection_state_receive_header *state =
3021 &conn->protocol.ctrl.state.receive_header;
3022
3023 assert(state->left_to_receive != 0);
3024
3025 ret = conn->sock->ops->recvmsg(conn->sock,
3026 reception_buffer->data + state->received,
3027 state->left_to_receive, MSG_DONTWAIT);
3028 if (ret < 0) {
3029 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3030 PERROR("Unable to receive control command header on sock %d",
3031 conn->sock->fd);
3032 status = RELAY_CONNECTION_STATUS_ERROR;
3033 }
3034 goto end;
3035 } else if (ret == 0) {
3036 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3037 status = RELAY_CONNECTION_STATUS_CLOSED;
3038 goto end;
3039 }
3040
3041 assert(ret > 0);
3042 assert(ret <= state->left_to_receive);
3043
3044 state->left_to_receive -= ret;
3045 state->received += ret;
3046
3047 if (state->left_to_receive > 0) {
3048 /*
3049 * Can't transition to the protocol's next state, wait to
3050 * receive the rest of the header.
3051 */
3052 DBG3("Partial reception of control connection protocol header (received %" PRIu64 " bytes, %" PRIu64 " bytes left to receive, fd = %i)",
3053 state->received, state->left_to_receive,
3054 conn->sock->fd);
3055 goto end;
3056 }
3057
3058 /* Transition to next state: receiving the command's payload. */
3059 conn->protocol.ctrl.state_id =
3060 CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD;
3061 memcpy(&header, reception_buffer->data, sizeof(header));
3062 header.circuit_id = be64toh(header.circuit_id);
3063 header.data_size = be64toh(header.data_size);
3064 header.cmd = be32toh(header.cmd);
3065 header.cmd_version = be32toh(header.cmd_version);
3066 memcpy(&conn->protocol.ctrl.state.receive_payload.header,
3067 &header, sizeof(header));
3068
3069 DBG("Done receiving control command header: fd = %i, cmd = %" PRIu32 ", cmd_version = %" PRIu32 ", payload size = %" PRIu64 " bytes",
3070 conn->sock->fd, header.cmd, header.cmd_version,
3071 header.data_size);
3072
3073 if (header.data_size > DEFAULT_NETWORK_RELAYD_CTRL_MAX_PAYLOAD_SIZE) {
3074 ERR("Command header indicates a payload (%" PRIu64 " bytes) that exceeds the maximal payload size allowed on a control connection.",
3075 header.data_size);
3076 status = RELAY_CONNECTION_STATUS_ERROR;
3077 goto end;
3078 }
3079
3080 conn->protocol.ctrl.state.receive_payload.left_to_receive =
3081 header.data_size;
3082 conn->protocol.ctrl.state.receive_payload.received = 0;
3083 ret = lttng_dynamic_buffer_set_size(reception_buffer,
3084 header.data_size);
3085 if (ret) {
3086 status = RELAY_CONNECTION_STATUS_ERROR;
3087 goto end;
3088 }
3089
3090 if (header.data_size == 0) {
3091 /*
3092 * Manually invoke the next state as the poll loop
3093 * will not wake-up to allow us to proceed further.
3094 */
3095 status = relay_process_control_receive_payload(conn);
3096 }
3097 end:
3098 return status;
3099 }
3100
3101 /*
3102 * Process the commands received on the control socket
3103 */
3104 static enum relay_connection_status relay_process_control(
3105 struct relay_connection *conn)
3106 {
3107 enum relay_connection_status status;
3108
3109 switch (conn->protocol.ctrl.state_id) {
3110 case CTRL_CONNECTION_STATE_RECEIVE_HEADER:
3111 status = relay_process_control_receive_header(conn);
3112 break;
3113 case CTRL_CONNECTION_STATE_RECEIVE_PAYLOAD:
3114 status = relay_process_control_receive_payload(conn);
3115 break;
3116 default:
3117 ERR("Unknown control connection protocol state encountered.");
3118 abort();
3119 }
3120
3121 return status;
3122 }
3123
3124 static enum relay_connection_status relay_process_data_receive_header(
3125 struct relay_connection *conn)
3126 {
3127 int ret;
3128 enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
3129 struct data_connection_state_receive_header *state =
3130 &conn->protocol.data.state.receive_header;
3131 struct lttcomm_relayd_data_hdr header;
3132 struct relay_stream *stream;
3133
3134 assert(state->left_to_receive != 0);
3135
3136 ret = conn->sock->ops->recvmsg(conn->sock,
3137 state->header_reception_buffer + state->received,
3138 state->left_to_receive, MSG_DONTWAIT);
3139 if (ret < 0) {
3140 if (errno != EAGAIN && errno != EWOULDBLOCK) {
3141 PERROR("Unable to receive data header on sock %d", conn->sock->fd);
3142 status = RELAY_CONNECTION_STATUS_ERROR;
3143 }
3144 goto end;
3145 } else if (ret == 0) {
3146 /* Orderly shutdown. Not necessary to print an error. */
3147 DBG("Socket %d performed an orderly shutdown (received EOF)", conn->sock->fd);
3148 status = RELAY_CONNECTION_STATUS_CLOSED;
3149 goto end;
3150 }
3151
3152 assert(ret > 0);
3153 assert(ret <= state->left_to_receive);
3154
3155 state->left_to_receive -= ret;
3156 state->received += ret;
3157
3158 if (state->left_to_receive > 0) {
3159 /*
3160 * Can't transition to the protocol's next state, wait to
3161 * receive the rest of the header.
3162 */
3163 DBG3("Partial reception of data connection header (received %" PRIu64 " bytes, %" PRIu64 " bytes