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