228b35cb26b4414854a6218bded70003905df28b
[lttng-tools.git] / src / common / relayd / relayd.c
1 /*
2 * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
3 *
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License, version 2 only, as
6 * published by the Free Software Foundation.
7 *
8 * This program is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
11 * more details.
12 *
13 * You should have received a copy of the GNU General Public License along with
14 * this program; if not, write to the Free Software Foundation, Inc., 51
15 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16 */
17
18 #define _LGPL_SOURCE
19 #include <assert.h>
20 #include <stdio.h>
21 #include <stdlib.h>
22 #include <string.h>
23 #include <sys/stat.h>
24 #include <inttypes.h>
25
26 #include <common/common.h>
27 #include <common/defaults.h>
28 #include <common/compat/endian.h>
29 #include <common/compat/string.h>
30 #include <common/sessiond-comm/relayd.h>
31 #include <common/index/ctf-index.h>
32 #include <common/trace-chunk.h>
33 #include <common/string-utils/format.h>
34
35 #include "relayd.h"
36
37 static
38 bool relayd_supports_chunks(const struct lttcomm_relayd_sock *sock)
39 {
40 if (sock->major > 2) {
41 return true;
42 } else if (sock->major == 2 && sock->minor >= 11) {
43 return true;
44 }
45 return false;
46 }
47
48 /*
49 * Send command. Fill up the header and append the data.
50 */
51 static int send_command(struct lttcomm_relayd_sock *rsock,
52 enum lttcomm_relayd_command cmd, const void *data, size_t size,
53 int flags)
54 {
55 int ret;
56 struct lttcomm_relayd_hdr header;
57 char *buf;
58 uint64_t buf_size = sizeof(header);
59
60 if (rsock->sock.fd < 0) {
61 return -ECONNRESET;
62 }
63
64 if (data) {
65 buf_size += size;
66 }
67
68 buf = zmalloc(buf_size);
69 if (buf == NULL) {
70 PERROR("zmalloc relayd send command buf");
71 ret = -1;
72 goto alloc_error;
73 }
74
75 memset(&header, 0, sizeof(header));
76 header.cmd = htobe32(cmd);
77 header.data_size = htobe64(size);
78
79 /* Zeroed for now since not used. */
80 header.cmd_version = 0;
81 header.circuit_id = 0;
82
83 /* Prepare buffer to send. */
84 memcpy(buf, &header, sizeof(header));
85 if (data) {
86 memcpy(buf + sizeof(header), data, size);
87 }
88
89 DBG3("Relayd sending command %d of size %" PRIu64, (int) cmd, buf_size);
90 ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags);
91 if (ret < 0) {
92 PERROR("Failed to send command %d of size %" PRIu64,
93 (int) cmd, buf_size);
94 ret = -errno;
95 goto error;
96 }
97 error:
98 free(buf);
99 alloc_error:
100 return ret;
101 }
102
103 /*
104 * Receive reply data on socket. This MUST be call after send_command or else
105 * could result in unexpected behavior(s).
106 */
107 static int recv_reply(struct lttcomm_relayd_sock *rsock, void *data, size_t size)
108 {
109 int ret;
110
111 if (rsock->sock.fd < 0) {
112 return -ECONNRESET;
113 }
114
115 DBG3("Relayd waiting for reply of size %zu", size);
116
117 ret = rsock->sock.ops->recvmsg(&rsock->sock, data, size, 0);
118 if (ret <= 0 || ret != size) {
119 if (ret == 0) {
120 /* Orderly shutdown. */
121 DBG("Socket %d has performed an orderly shutdown", rsock->sock.fd);
122 } else {
123 DBG("Receiving reply failed on sock %d for size %zu with ret %d",
124 rsock->sock.fd, size, ret);
125 }
126 /* Always return -1 here and the caller can use errno. */
127 ret = -1;
128 goto error;
129 }
130
131 error:
132 return ret;
133 }
134
135 /*
136 * Starting from 2.11, RELAYD_CREATE_SESSION payload (session_name,
137 * hostname, and base_path) have no length restriction on the sender side.
138 * Length for both payloads is stored in the msg struct. A new dynamic size
139 * payload size is introduced.
140 */
141 static int relayd_create_session_2_11(struct lttcomm_relayd_sock *rsock,
142 const char *session_name, const char *hostname,
143 const char *base_path, int session_live_timer,
144 unsigned int snapshot, uint64_t sessiond_session_id,
145 const lttng_uuid sessiond_uuid, const uint64_t *current_chunk_id,
146 time_t creation_time, bool session_name_contains_creation_time,
147 struct lttcomm_relayd_create_session_reply_2_11 *reply,
148 char *output_path)
149 {
150 int ret;
151 struct lttcomm_relayd_create_session_2_11 *msg = NULL;
152 size_t session_name_len;
153 size_t hostname_len;
154 size_t base_path_len;
155 size_t msg_length;
156 char *dst;
157
158 if (!base_path) {
159 base_path = "";
160 }
161 /* The three names are sent with a '\0' delimiter between them. */
162 session_name_len = strlen(session_name) + 1;
163 hostname_len = strlen(hostname) + 1;
164 base_path_len = strlen(base_path) + 1;
165
166 msg_length = sizeof(*msg) + session_name_len + hostname_len + base_path_len;
167 msg = zmalloc(msg_length);
168 if (!msg) {
169 PERROR("zmalloc create_session_2_11 command message");
170 ret = -1;
171 goto error;
172 }
173
174 assert(session_name_len <= UINT32_MAX);
175 msg->session_name_len = htobe32(session_name_len);
176
177 assert(hostname_len <= UINT32_MAX);
178 msg->hostname_len = htobe32(hostname_len);
179
180 assert(base_path_len <= UINT32_MAX);
181 msg->base_path_len = htobe32(base_path_len);
182
183 dst = msg->names;
184 if (lttng_strncpy(dst, session_name, session_name_len)) {
185 ret = -1;
186 goto error;
187 }
188 dst += session_name_len;
189 if (lttng_strncpy(dst, hostname, hostname_len)) {
190 ret = -1;
191 goto error;
192 }
193 dst += hostname_len;
194 if (lttng_strncpy(dst, base_path, base_path_len)) {
195 ret = -1;
196 goto error;
197 }
198
199 msg->live_timer = htobe32(session_live_timer);
200 msg->snapshot = !!snapshot;
201
202 lttng_uuid_copy(msg->sessiond_uuid, sessiond_uuid);
203 msg->session_id = htobe64(sessiond_session_id);
204 msg->session_name_contains_creation_time = session_name_contains_creation_time;
205 if (current_chunk_id) {
206 LTTNG_OPTIONAL_SET(&msg->current_chunk_id,
207 htobe64(*current_chunk_id));
208 }
209
210 msg->creation_time = htobe64((uint64_t) creation_time);
211
212 /* Send command */
213 ret = send_command(rsock, RELAYD_CREATE_SESSION, msg, msg_length, 0);
214 if (ret < 0) {
215 goto error;
216 }
217 /* Receive response */
218 ret = recv_reply(rsock, reply, sizeof(*reply));
219 if (ret < 0) {
220 goto error;
221 }
222 reply->generic.session_id = be64toh(reply->generic.session_id);
223 reply->generic.ret_code = be32toh(reply->generic.ret_code);
224 reply->output_path_length = be32toh(reply->output_path_length);
225 if (reply->output_path_length >= LTTNG_PATH_MAX) {
226 ERR("Invalid session output path length in reply (%" PRIu32 " bytes) exceeds maximal allowed length (%d bytes)",
227 reply->output_path_length, LTTNG_PATH_MAX);
228 ret = -1;
229 goto error;
230 }
231 ret = recv_reply(rsock, output_path, reply->output_path_length);
232 if (ret < 0) {
233 goto error;
234 }
235 error:
236 free(msg);
237 return ret;
238 }
239 /*
240 * From 2.4 to 2.10, RELAYD_CREATE_SESSION takes additional parameters to
241 * support the live reading capability.
242 */
243 static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock,
244 const char *session_name, const char *hostname,
245 int session_live_timer, unsigned int snapshot,
246 struct lttcomm_relayd_status_session *reply)
247 {
248 int ret;
249 struct lttcomm_relayd_create_session_2_4 msg;
250
251 if (lttng_strncpy(msg.session_name, session_name,
252 sizeof(msg.session_name))) {
253 ret = -1;
254 goto error;
255 }
256 if (lttng_strncpy(msg.hostname, hostname, sizeof(msg.hostname))) {
257 ret = -1;
258 goto error;
259 }
260 msg.live_timer = htobe32(session_live_timer);
261 msg.snapshot = htobe32(snapshot);
262
263 /* Send command */
264 ret = send_command(rsock, RELAYD_CREATE_SESSION, &msg, sizeof(msg), 0);
265 if (ret < 0) {
266 goto error;
267 }
268
269 /* Receive response */
270 ret = recv_reply(rsock, reply, sizeof(*reply));
271 if (ret < 0) {
272 goto error;
273 }
274 reply->session_id = be64toh(reply->session_id);
275 reply->ret_code = be32toh(reply->ret_code);
276 error:
277 return ret;
278 }
279
280 /*
281 * RELAYD_CREATE_SESSION from 2.1 to 2.3.
282 */
283 static int relayd_create_session_2_1(struct lttcomm_relayd_sock *rsock,
284 struct lttcomm_relayd_status_session *reply)
285 {
286 int ret;
287
288 /* Send command */
289 ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0);
290 if (ret < 0) {
291 goto error;
292 }
293
294 /* Receive response */
295 ret = recv_reply(rsock, reply, sizeof(*reply));
296 if (ret < 0) {
297 goto error;
298 }
299 reply->session_id = be64toh(reply->session_id);
300 reply->ret_code = be32toh(reply->ret_code);
301 error:
302 return ret;
303 }
304
305 /*
306 * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and
307 * set session_id of the relayd if we have a successful reply from the relayd.
308 *
309 * On success, return 0 else a negative value which is either an errno error or
310 * a lttng error code from the relayd.
311 */
312 int relayd_create_session(struct lttcomm_relayd_sock *rsock,
313 uint64_t *relayd_session_id,
314 const char *session_name, const char *hostname,
315 const char *base_path, int session_live_timer,
316 unsigned int snapshot, uint64_t sessiond_session_id,
317 const lttng_uuid sessiond_uuid,
318 const uint64_t *current_chunk_id,
319 time_t creation_time, bool session_name_contains_creation_time,
320 char *output_path)
321 {
322 int ret;
323 struct lttcomm_relayd_create_session_reply_2_11 reply = {};
324
325 assert(rsock);
326 assert(relayd_session_id);
327
328 DBG("Relayd create session");
329
330 if (rsock->minor < 4) {
331 /* From 2.1 to 2.3 */
332 ret = relayd_create_session_2_1(rsock, &reply.generic);
333 } else if (rsock->minor >= 4 && rsock->minor < 11) {
334 /* From 2.4 to 2.10 */
335 ret = relayd_create_session_2_4(rsock, session_name,
336 hostname, session_live_timer, snapshot,
337 &reply.generic);
338 } else {
339 /* From 2.11 to ... */
340 ret = relayd_create_session_2_11(rsock, session_name,
341 hostname, base_path, session_live_timer, snapshot,
342 sessiond_session_id, sessiond_uuid,
343 current_chunk_id, creation_time,
344 session_name_contains_creation_time,
345 &reply, output_path);
346 }
347
348 if (ret < 0) {
349 goto error;
350 }
351
352 /* Return session id or negative ret code. */
353 if (reply.generic.ret_code != LTTNG_OK) {
354 ret = -1;
355 ERR("Relayd create session replied error %d",
356 reply.generic.ret_code);
357 goto error;
358 } else {
359 ret = 0;
360 *relayd_session_id = reply.generic.session_id;
361 }
362
363 DBG("Relayd session created with id %" PRIu64, reply.generic.session_id);
364
365 error:
366 return ret;
367 }
368
369 static int relayd_add_stream_2_1(struct lttcomm_relayd_sock *rsock,
370 const char *channel_name, const char *pathname)
371 {
372 int ret;
373 struct lttcomm_relayd_add_stream msg;
374
375 memset(&msg, 0, sizeof(msg));
376 if (lttng_strncpy(msg.channel_name, channel_name,
377 sizeof(msg.channel_name))) {
378 ret = -1;
379 goto error;
380 }
381
382 if (lttng_strncpy(msg.pathname, pathname,
383 sizeof(msg.pathname))) {
384 ret = -1;
385 goto error;
386 }
387
388 /* Send command */
389 ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
390 if (ret < 0) {
391 ret = -1;
392 goto error;
393 }
394 ret = 0;
395 error:
396 return ret;
397 }
398
399 static int relayd_add_stream_2_2(struct lttcomm_relayd_sock *rsock,
400 const char *channel_name, const char *pathname,
401 uint64_t tracefile_size, uint64_t tracefile_count)
402 {
403 int ret;
404 struct lttcomm_relayd_add_stream_2_2 msg;
405
406 memset(&msg, 0, sizeof(msg));
407 /* Compat with relayd 2.2 to 2.10 */
408 if (lttng_strncpy(msg.channel_name, channel_name,
409 sizeof(msg.channel_name))) {
410 ret = -1;
411 goto error;
412 }
413 if (lttng_strncpy(msg.pathname, pathname,
414 sizeof(msg.pathname))) {
415 ret = -1;
416 goto error;
417 }
418 msg.tracefile_size = htobe64(tracefile_size);
419 msg.tracefile_count = htobe64(tracefile_count);
420
421 /* Send command */
422 ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
423 if (ret < 0) {
424 goto error;
425 }
426 ret = 0;
427 error:
428 return ret;
429 }
430
431 static int relayd_add_stream_2_11(struct lttcomm_relayd_sock *rsock,
432 const char *channel_name, const char *pathname,
433 uint64_t tracefile_size, uint64_t tracefile_count,
434 uint64_t trace_archive_id)
435 {
436 int ret;
437 struct lttcomm_relayd_add_stream_2_11 *msg = NULL;
438 size_t channel_name_len;
439 size_t pathname_len;
440 size_t msg_length;
441
442 /* The two names are sent with a '\0' delimiter between them. */
443 channel_name_len = strlen(channel_name) + 1;
444 pathname_len = strlen(pathname) + 1;
445
446 msg_length = sizeof(*msg) + channel_name_len + pathname_len;
447 msg = zmalloc(msg_length);
448 if (!msg) {
449 PERROR("zmalloc add_stream_2_11 command message");
450 ret = -1;
451 goto error;
452 }
453
454 assert(channel_name_len <= UINT32_MAX);
455 msg->channel_name_len = htobe32(channel_name_len);
456
457 assert(pathname_len <= UINT32_MAX);
458 msg->pathname_len = htobe32(pathname_len);
459
460 if (lttng_strncpy(msg->names, channel_name, channel_name_len)) {
461 ret = -1;
462 goto error;
463 }
464 if (lttng_strncpy(msg->names + channel_name_len, pathname, pathname_len)) {
465 ret = -1;
466 goto error;
467 }
468
469 msg->tracefile_size = htobe64(tracefile_size);
470 msg->tracefile_count = htobe64(tracefile_count);
471 msg->trace_chunk_id = htobe64(trace_archive_id);
472
473 /* Send command */
474 ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) msg, msg_length, 0);
475 if (ret < 0) {
476 goto error;
477 }
478 ret = 0;
479 error:
480 free(msg);
481 return ret;
482 }
483
484 /*
485 * Add stream on the relayd and assign stream handle to the stream_id argument.
486 *
487 * Chunks are not supported by relayd prior to 2.11, but are used to
488 * internally between session daemon and consumer daemon to keep track
489 * of the channel and stream output path.
490 *
491 * On success return 0 else return ret_code negative value.
492 */
493 int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name,
494 const char *domain_name, const char *_pathname, uint64_t *stream_id,
495 uint64_t tracefile_size, uint64_t tracefile_count,
496 struct lttng_trace_chunk *trace_chunk)
497 {
498 int ret;
499 struct lttcomm_relayd_status_stream reply;
500 char pathname[RELAYD_COMM_LTTNG_PATH_MAX];
501 char *separator;
502
503 /* Code flow error. Safety net. */
504 assert(rsock);
505 assert(channel_name);
506 assert(domain_name);
507 assert(_pathname);
508 assert(trace_chunk);
509
510 DBG("Relayd adding stream for channel name %s", channel_name);
511
512 if (_pathname[0] == '\0') {
513 separator = "";
514 } else {
515 separator = "/";
516 }
517 ret = snprintf(pathname, RELAYD_COMM_LTTNG_PATH_MAX, "%s%s%s",
518 domain_name, separator, _pathname);
519 if (ret <= 0 || ret >= RELAYD_COMM_LTTNG_PATH_MAX) {
520 ERR("stream path too long.");
521 ret = -1;
522 goto error;
523 }
524
525 /* Compat with relayd 2.1 */
526 if (rsock->minor == 1) {
527 /* For 2.1 */
528 ret = relayd_add_stream_2_1(rsock, channel_name, pathname);
529
530 } else if (rsock->minor > 1 && rsock->minor < 11) {
531 /* From 2.2 to 2.10 */
532 ret = relayd_add_stream_2_2(rsock, channel_name, pathname,
533 tracefile_size, tracefile_count);
534 } else {
535 enum lttng_trace_chunk_status chunk_status;
536 uint64_t chunk_id;
537
538 chunk_status = lttng_trace_chunk_get_id(trace_chunk,
539 &chunk_id);
540 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
541
542 /* From 2.11 to ...*/
543 ret = relayd_add_stream_2_11(rsock, channel_name, pathname,
544 tracefile_size, tracefile_count,
545 chunk_id);
546 }
547
548 if (ret) {
549 ret = -1;
550 goto error;
551 }
552
553 /* Waiting for reply */
554 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
555 if (ret < 0) {
556 goto error;
557 }
558
559 /* Back to host bytes order. */
560 reply.handle = be64toh(reply.handle);
561 reply.ret_code = be32toh(reply.ret_code);
562
563 /* Return session id or negative ret code. */
564 if (reply.ret_code != LTTNG_OK) {
565 ret = -1;
566 ERR("Relayd add stream replied error %d", reply.ret_code);
567 } else {
568 /* Success */
569 ret = 0;
570 *stream_id = reply.handle;
571 }
572
573 DBG("Relayd stream added successfully with handle %" PRIu64,
574 reply.handle);
575
576 error:
577 return ret;
578 }
579
580 /*
581 * Inform the relay that all the streams for the current channel has been sent.
582 *
583 * On success return 0 else return ret_code negative value.
584 */
585 int relayd_streams_sent(struct lttcomm_relayd_sock *rsock)
586 {
587 int ret;
588 struct lttcomm_relayd_generic_reply reply;
589
590 /* Code flow error. Safety net. */
591 assert(rsock);
592
593 DBG("Relayd sending streams sent.");
594
595 /* This feature was introduced in 2.4, ignore it for earlier versions. */
596 if (rsock->minor < 4) {
597 ret = 0;
598 goto end;
599 }
600
601 /* Send command */
602 ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0);
603 if (ret < 0) {
604 goto error;
605 }
606
607 /* Waiting for reply */
608 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
609 if (ret < 0) {
610 goto error;
611 }
612
613 /* Back to host bytes order. */
614 reply.ret_code = be32toh(reply.ret_code);
615
616 /* Return session id or negative ret code. */
617 if (reply.ret_code != LTTNG_OK) {
618 ret = -1;
619 ERR("Relayd streams sent replied error %d", reply.ret_code);
620 goto error;
621 } else {
622 /* Success */
623 ret = 0;
624 }
625
626 DBG("Relayd streams sent success");
627
628 error:
629 end:
630 return ret;
631 }
632
633 /*
634 * Check version numbers on the relayd.
635 * If major versions are compatible, we assign minor_to_use to the
636 * minor version of the procotol we are going to use for this session.
637 *
638 * Return 0 if the two daemons are compatible, LTTNG_ERR_RELAYD_VERSION_FAIL
639 * otherwise, or a negative value on network errors.
640 */
641 int relayd_version_check(struct lttcomm_relayd_sock *rsock)
642 {
643 int ret;
644 struct lttcomm_relayd_version msg;
645
646 /* Code flow error. Safety net. */
647 assert(rsock);
648
649 DBG("Relayd version check for major.minor %u.%u", rsock->major,
650 rsock->minor);
651
652 memset(&msg, 0, sizeof(msg));
653 /* Prepare network byte order before transmission. */
654 msg.major = htobe32(rsock->major);
655 msg.minor = htobe32(rsock->minor);
656
657 /* Send command */
658 ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0);
659 if (ret < 0) {
660 goto error;
661 }
662
663 /* Receive response */
664 ret = recv_reply(rsock, (void *) &msg, sizeof(msg));
665 if (ret < 0) {
666 goto error;
667 }
668
669 /* Set back to host bytes order */
670 msg.major = be32toh(msg.major);
671 msg.minor = be32toh(msg.minor);
672
673 /*
674 * Only validate the major version. If the other side is higher,
675 * communication is not possible. Only major version equal can talk to each
676 * other. If the minor version differs, the lowest version is used by both
677 * sides.
678 */
679 if (msg.major != rsock->major) {
680 /* Not compatible */
681 ret = LTTNG_ERR_RELAYD_VERSION_FAIL;
682 DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)",
683 msg.major, rsock->major);
684 goto error;
685 }
686
687 /*
688 * If the relayd's minor version is higher, it will adapt to our version so
689 * we can continue to use the latest relayd communication data structure.
690 * If the received minor version is higher, the relayd should adapt to us.
691 */
692 if (rsock->minor > msg.minor) {
693 rsock->minor = msg.minor;
694 }
695
696 /* Version number compatible */
697 DBG2("Relayd version is compatible, using protocol version %u.%u",
698 rsock->major, rsock->minor);
699 ret = 0;
700
701 error:
702 return ret;
703 }
704
705 /*
706 * Add stream on the relayd and assign stream handle to the stream_id argument.
707 *
708 * On success return 0 else return ret_code negative value.
709 */
710 int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len)
711 {
712 int ret;
713
714 /* Code flow error. Safety net. */
715 assert(rsock);
716
717 DBG("Relayd sending metadata of size %zu", len);
718
719 /* Send command */
720 ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0);
721 if (ret < 0) {
722 goto error;
723 }
724
725 DBG2("Relayd metadata added successfully");
726
727 /*
728 * After that call, the metadata data MUST be sent to the relayd so the
729 * receive size on the other end matches the len of the metadata packet
730 * header. This is why we don't wait for a reply here.
731 */
732
733 error:
734 return ret;
735 }
736
737 /*
738 * Connect to relay daemon with an allocated lttcomm_relayd_sock.
739 */
740 int relayd_connect(struct lttcomm_relayd_sock *rsock)
741 {
742 /* Code flow error. Safety net. */
743 assert(rsock);
744
745 if (!rsock->sock.ops) {
746 /*
747 * Attempting a connect on a non-initialized socket.
748 */
749 return -ECONNRESET;
750 }
751
752 DBG3("Relayd connect ...");
753
754 return rsock->sock.ops->connect(&rsock->sock);
755 }
756
757 /*
758 * Close relayd socket with an allocated lttcomm_relayd_sock.
759 *
760 * If no socket operations are found, simply return 0 meaning that everything
761 * is fine. Without operations, the socket can not possibly be opened or used.
762 * This is possible if the socket was allocated but not created. However, the
763 * caller could simply use it to store a valid file descriptor for instance
764 * passed over a Unix socket and call this to cleanup but still without a valid
765 * ops pointer.
766 *
767 * Return the close returned value. On error, a negative value is usually
768 * returned back from close(2).
769 */
770 int relayd_close(struct lttcomm_relayd_sock *rsock)
771 {
772 int ret;
773
774 /* Code flow error. Safety net. */
775 assert(rsock);
776
777 /* An invalid fd is fine, return success. */
778 if (rsock->sock.fd < 0) {
779 ret = 0;
780 goto end;
781 }
782
783 DBG3("Relayd closing socket %d", rsock->sock.fd);
784
785 if (rsock->sock.ops) {
786 ret = rsock->sock.ops->close(&rsock->sock);
787 } else {
788 /* Default call if no specific ops found. */
789 ret = close(rsock->sock.fd);
790 if (ret < 0) {
791 PERROR("relayd_close default close");
792 }
793 }
794 rsock->sock.fd = -1;
795
796 end:
797 return ret;
798 }
799
800 /*
801 * Send data header structure to the relayd.
802 */
803 int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock,
804 struct lttcomm_relayd_data_hdr *hdr, size_t size)
805 {
806 int ret;
807
808 /* Code flow error. Safety net. */
809 assert(rsock);
810 assert(hdr);
811
812 if (rsock->sock.fd < 0) {
813 return -ECONNRESET;
814 }
815
816 DBG3("Relayd sending data header of size %zu", size);
817
818 /* Again, safety net */
819 if (size == 0) {
820 size = sizeof(struct lttcomm_relayd_data_hdr);
821 }
822
823 /* Only send data header. */
824 ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0);
825 if (ret < 0) {
826 ret = -errno;
827 goto error;
828 }
829
830 /*
831 * The data MUST be sent right after that command for the receive on the
832 * other end to match the size in the header.
833 */
834
835 error:
836 return ret;
837 }
838
839 /*
840 * Send close stream command to the relayd.
841 */
842 int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
843 uint64_t last_net_seq_num)
844 {
845 int ret;
846 struct lttcomm_relayd_close_stream msg;
847 struct lttcomm_relayd_generic_reply reply;
848
849 /* Code flow error. Safety net. */
850 assert(rsock);
851
852 DBG("Relayd closing stream id %" PRIu64, stream_id);
853
854 memset(&msg, 0, sizeof(msg));
855 msg.stream_id = htobe64(stream_id);
856 msg.last_net_seq_num = htobe64(last_net_seq_num);
857
858 /* Send command */
859 ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
860 if (ret < 0) {
861 goto error;
862 }
863
864 /* Receive response */
865 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
866 if (ret < 0) {
867 goto error;
868 }
869
870 reply.ret_code = be32toh(reply.ret_code);
871
872 /* Return session id or negative ret code. */
873 if (reply.ret_code != LTTNG_OK) {
874 ret = -1;
875 ERR("Relayd close stream replied error %d", reply.ret_code);
876 } else {
877 /* Success */
878 ret = 0;
879 }
880
881 DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
882
883 error:
884 return ret;
885 }
886
887 /*
888 * Check for data availability for a given stream id.
889 *
890 * Return 0 if NOT pending, 1 if so and a negative value on error.
891 */
892 int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
893 uint64_t last_net_seq_num)
894 {
895 int ret;
896 struct lttcomm_relayd_data_pending msg;
897 struct lttcomm_relayd_generic_reply reply;
898
899 /* Code flow error. Safety net. */
900 assert(rsock);
901
902 DBG("Relayd data pending for stream id %" PRIu64, stream_id);
903
904 memset(&msg, 0, sizeof(msg));
905 msg.stream_id = htobe64(stream_id);
906 msg.last_net_seq_num = htobe64(last_net_seq_num);
907
908 /* Send command */
909 ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg,
910 sizeof(msg), 0);
911 if (ret < 0) {
912 goto error;
913 }
914
915 /* Receive response */
916 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
917 if (ret < 0) {
918 goto error;
919 }
920
921 reply.ret_code = be32toh(reply.ret_code);
922
923 /* Return session id or negative ret code. */
924 if (reply.ret_code >= LTTNG_OK) {
925 ERR("Relayd data pending replied error %d", reply.ret_code);
926 }
927
928 /* At this point, the ret code is either 1 or 0 */
929 ret = reply.ret_code;
930
931 DBG("Relayd data is %s pending for stream id %" PRIu64,
932 ret == 1 ? "" : "NOT", stream_id);
933
934 error:
935 return ret;
936 }
937
938 /*
939 * Check on the relayd side for a quiescent state on the control socket.
940 */
941 int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock,
942 uint64_t metadata_stream_id)
943 {
944 int ret;
945 struct lttcomm_relayd_quiescent_control msg;
946 struct lttcomm_relayd_generic_reply reply;
947
948 /* Code flow error. Safety net. */
949 assert(rsock);
950
951 DBG("Relayd checking quiescent control state");
952
953 memset(&msg, 0, sizeof(msg));
954 msg.stream_id = htobe64(metadata_stream_id);
955
956 /* Send command */
957 ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0);
958 if (ret < 0) {
959 goto error;
960 }
961
962 /* Receive response */
963 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
964 if (ret < 0) {
965 goto error;
966 }
967
968 reply.ret_code = be32toh(reply.ret_code);
969
970 /* Return session id or negative ret code. */
971 if (reply.ret_code != LTTNG_OK) {
972 ret = -1;
973 ERR("Relayd quiescent control replied error %d", reply.ret_code);
974 goto error;
975 }
976
977 /* Control socket is quiescent */
978 return 0;
979
980 error:
981 return ret;
982 }
983
984 /*
985 * Begin a data pending command for a specific session id.
986 */
987 int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id)
988 {
989 int ret;
990 struct lttcomm_relayd_begin_data_pending msg;
991 struct lttcomm_relayd_generic_reply reply;
992
993 /* Code flow error. Safety net. */
994 assert(rsock);
995
996 DBG("Relayd begin data pending");
997
998 memset(&msg, 0, sizeof(msg));
999 msg.session_id = htobe64(id);
1000
1001 /* Send command */
1002 ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0);
1003 if (ret < 0) {
1004 goto error;
1005 }
1006
1007 /* Receive response */
1008 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1009 if (ret < 0) {
1010 goto error;
1011 }
1012
1013 reply.ret_code = be32toh(reply.ret_code);
1014
1015 /* Return session id or negative ret code. */
1016 if (reply.ret_code != LTTNG_OK) {
1017 ret = -1;
1018 ERR("Relayd begin data pending replied error %d", reply.ret_code);
1019 goto error;
1020 }
1021
1022 return 0;
1023
1024 error:
1025 return ret;
1026 }
1027
1028 /*
1029 * End a data pending command for a specific session id.
1030 *
1031 * Return 0 on success and set is_data_inflight to 0 if no data is being
1032 * streamed or 1 if it is the case.
1033 */
1034 int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id,
1035 unsigned int *is_data_inflight)
1036 {
1037 int ret, recv_ret;
1038 struct lttcomm_relayd_end_data_pending msg;
1039 struct lttcomm_relayd_generic_reply reply;
1040
1041 /* Code flow error. Safety net. */
1042 assert(rsock);
1043
1044 DBG("Relayd end data pending");
1045
1046 memset(&msg, 0, sizeof(msg));
1047 msg.session_id = htobe64(id);
1048
1049 /* Send command */
1050 ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0);
1051 if (ret < 0) {
1052 goto error;
1053 }
1054
1055 /* Receive response */
1056 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1057 if (ret < 0) {
1058 goto error;
1059 }
1060
1061 recv_ret = be32toh(reply.ret_code);
1062 if (recv_ret < 0) {
1063 ret = recv_ret;
1064 goto error;
1065 }
1066
1067 *is_data_inflight = recv_ret;
1068
1069 DBG("Relayd end data pending is data inflight: %d", recv_ret);
1070
1071 return 0;
1072
1073 error:
1074 return ret;
1075 }
1076
1077 /*
1078 * Send index to the relayd.
1079 */
1080 int relayd_send_index(struct lttcomm_relayd_sock *rsock,
1081 struct ctf_packet_index *index, uint64_t relay_stream_id,
1082 uint64_t net_seq_num)
1083 {
1084 int ret;
1085 struct lttcomm_relayd_index msg;
1086 struct lttcomm_relayd_generic_reply reply;
1087
1088 /* Code flow error. Safety net. */
1089 assert(rsock);
1090
1091 if (rsock->minor < 4) {
1092 DBG("Not sending indexes before protocol 2.4");
1093 ret = 0;
1094 goto error;
1095 }
1096
1097 DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id);
1098
1099 memset(&msg, 0, sizeof(msg));
1100 msg.relay_stream_id = htobe64(relay_stream_id);
1101 msg.net_seq_num = htobe64(net_seq_num);
1102
1103 /* The index is already in big endian. */
1104 msg.packet_size = index->packet_size;
1105 msg.content_size = index->content_size;
1106 msg.timestamp_begin = index->timestamp_begin;
1107 msg.timestamp_end = index->timestamp_end;
1108 msg.events_discarded = index->events_discarded;
1109 msg.stream_id = index->stream_id;
1110
1111 if (rsock->minor >= 8) {
1112 msg.stream_instance_id = index->stream_instance_id;
1113 msg.packet_seq_num = index->packet_seq_num;
1114 }
1115
1116 /* Send command */
1117 ret = send_command(rsock, RELAYD_SEND_INDEX, &msg,
1118 lttcomm_relayd_index_len(lttng_to_index_major(rsock->major,
1119 rsock->minor),
1120 lttng_to_index_minor(rsock->major, rsock->minor)),
1121 0);
1122 if (ret < 0) {
1123 goto error;
1124 }
1125
1126 /* Receive response */
1127 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1128 if (ret < 0) {
1129 goto error;
1130 }
1131
1132 reply.ret_code = be32toh(reply.ret_code);
1133
1134 /* Return session id or negative ret code. */
1135 if (reply.ret_code != LTTNG_OK) {
1136 ret = -1;
1137 ERR("Relayd send index replied error %d", reply.ret_code);
1138 } else {
1139 /* Success */
1140 ret = 0;
1141 }
1142
1143 error:
1144 return ret;
1145 }
1146
1147 /*
1148 * Ask the relay to reset the metadata trace file (regeneration).
1149 */
1150 int relayd_reset_metadata(struct lttcomm_relayd_sock *rsock,
1151 uint64_t stream_id, uint64_t version)
1152 {
1153 int ret;
1154 struct lttcomm_relayd_reset_metadata msg;
1155 struct lttcomm_relayd_generic_reply reply;
1156
1157 /* Code flow error. Safety net. */
1158 assert(rsock);
1159
1160 /* Should have been prevented by the sessiond. */
1161 if (rsock->minor < 8) {
1162 ERR("Metadata regeneration unsupported before 2.8");
1163 ret = -1;
1164 goto error;
1165 }
1166
1167 DBG("Relayd reset metadata stream id %" PRIu64, stream_id);
1168
1169 memset(&msg, 0, sizeof(msg));
1170 msg.stream_id = htobe64(stream_id);
1171 msg.version = htobe64(version);
1172
1173 /* Send command */
1174 ret = send_command(rsock, RELAYD_RESET_METADATA, (void *) &msg, sizeof(msg), 0);
1175 if (ret < 0) {
1176 goto error;
1177 }
1178
1179 /* Receive response */
1180 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
1181 if (ret < 0) {
1182 goto error;
1183 }
1184
1185 reply.ret_code = be32toh(reply.ret_code);
1186
1187 /* Return session id or negative ret code. */
1188 if (reply.ret_code != LTTNG_OK) {
1189 ret = -1;
1190 ERR("Relayd reset metadata replied error %d", reply.ret_code);
1191 } else {
1192 /* Success */
1193 ret = 0;
1194 }
1195
1196 DBG("Relayd reset metadata stream id %" PRIu64 " successfully", stream_id);
1197
1198 error:
1199 return ret;
1200 }
1201
1202 int relayd_rotate_streams(struct lttcomm_relayd_sock *sock,
1203 unsigned int stream_count, const uint64_t *new_chunk_id,
1204 const struct relayd_stream_rotation_position *positions)
1205 {
1206 int ret;
1207 unsigned int i;
1208 struct lttng_dynamic_buffer payload;
1209 struct lttcomm_relayd_generic_reply reply = {};
1210 const struct lttcomm_relayd_rotate_streams msg = {
1211 .stream_count = htobe32((uint32_t) stream_count),
1212 .new_chunk_id = (typeof(msg.new_chunk_id)) {
1213 .is_set = !!new_chunk_id,
1214 .value = htobe64(new_chunk_id ? *new_chunk_id : 0),
1215 },
1216 };
1217 char new_chunk_id_buf[MAX_INT_DEC_LEN(*new_chunk_id)] = {};
1218 const char *new_chunk_id_str;
1219
1220 if (!relayd_supports_chunks(sock)) {
1221 DBG("Refusing to rotate remote streams: relayd does not support chunks");
1222 return 0;
1223 }
1224
1225 lttng_dynamic_buffer_init(&payload);
1226
1227 /* Code flow error. Safety net. */
1228 assert(sock);
1229
1230 if (new_chunk_id) {
1231 ret = snprintf(new_chunk_id_buf, sizeof(new_chunk_id_buf),
1232 "%" PRIu64, *new_chunk_id);
1233 if (ret == -1 || ret >= sizeof(new_chunk_id_buf)) {
1234 new_chunk_id_str = "formatting error";
1235 } else {
1236 new_chunk_id_str = new_chunk_id_buf;
1237 }
1238 } else {
1239 new_chunk_id_str = "none";
1240 }
1241
1242 DBG("Preparing \"rotate streams\" command payload: new_chunk_id = %s, stream_count = %u",
1243 new_chunk_id_str, stream_count);
1244
1245 ret = lttng_dynamic_buffer_append(&payload, &msg, sizeof(msg));
1246 if (ret) {
1247 ERR("Failed to allocate \"rotate streams\" command payload");
1248 goto error;
1249 }
1250
1251 for (i = 0; i < stream_count; i++) {
1252 const struct relayd_stream_rotation_position *position =
1253 &positions[i];
1254 const struct lttcomm_relayd_stream_rotation_position comm_position = {
1255 .stream_id = htobe64(position->stream_id),
1256 .rotate_at_seq_num = htobe64(
1257 position->rotate_at_seq_num),
1258 };
1259
1260 DBG("Rotate stream %" PRIu64 "at sequence number %" PRIu64,
1261 position->stream_id,
1262 position->rotate_at_seq_num);
1263 ret = lttng_dynamic_buffer_append(&payload, &comm_position,
1264 sizeof(comm_position));
1265 if (ret) {
1266 ERR("Failed to allocate \"rotate streams\" command payload");
1267 goto error;
1268 }
1269 }
1270
1271 /* Send command. */
1272 ret = send_command(sock, RELAYD_ROTATE_STREAMS, payload.data,
1273 payload.size, 0);
1274 if (ret < 0) {
1275 ERR("Failed to send \"rotate stream\" command");
1276 goto error;
1277 }
1278
1279 /* Receive response. */
1280 ret = recv_reply(sock, &reply, sizeof(reply));
1281 if (ret < 0) {
1282 ERR("Failed to receive \"rotate streams\" command reply");
1283 goto error;
1284 }
1285
1286 reply.ret_code = be32toh(reply.ret_code);
1287 if (reply.ret_code != LTTNG_OK) {
1288 ret = -1;
1289 ERR("Relayd rotate streams replied error %d", reply.ret_code);
1290 } else {
1291 /* Success. */
1292 ret = 0;
1293 DBG("Relayd rotated streams successfully");
1294 }
1295
1296 error:
1297 lttng_dynamic_buffer_reset(&payload);
1298 return ret;
1299 }
1300
1301 int relayd_create_trace_chunk(struct lttcomm_relayd_sock *sock,
1302 struct lttng_trace_chunk *chunk)
1303 {
1304 int ret = 0;
1305 enum lttng_trace_chunk_status status;
1306 struct lttcomm_relayd_create_trace_chunk msg = {};
1307 struct lttcomm_relayd_generic_reply reply = {};
1308 struct lttng_dynamic_buffer payload;
1309 uint64_t chunk_id;
1310 time_t creation_timestamp;
1311 const char *chunk_name;
1312 size_t chunk_name_length;
1313 bool overridden_name;
1314
1315 lttng_dynamic_buffer_init(&payload);
1316
1317 if (!relayd_supports_chunks(sock)) {
1318 DBG("Refusing to create remote trace chunk: relayd does not support chunks");
1319 goto end;
1320 }
1321
1322 status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1323 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1324 ret = -1;
1325 goto end;
1326 }
1327
1328 status = lttng_trace_chunk_get_creation_timestamp(
1329 chunk, &creation_timestamp);
1330 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1331 ret = -1;
1332 goto end;
1333 }
1334
1335 status = lttng_trace_chunk_get_name(
1336 chunk, &chunk_name, &overridden_name);
1337 if (status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1338 status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1339 ret = -1;
1340 goto end;
1341 }
1342
1343 chunk_name_length = overridden_name ? (strlen(chunk_name) + 1) : 0;
1344 msg = (typeof(msg)){
1345 .chunk_id = htobe64(chunk_id),
1346 .creation_timestamp = htobe64((uint64_t) creation_timestamp),
1347 .override_name_length = htobe32((uint32_t) chunk_name_length),
1348 };
1349
1350 ret = lttng_dynamic_buffer_append(&payload, &msg, sizeof(msg));
1351 if (ret) {
1352 goto end;
1353 }
1354 if (chunk_name_length) {
1355 ret = lttng_dynamic_buffer_append(
1356 &payload, chunk_name, chunk_name_length);
1357 if (ret) {
1358 goto end;
1359 }
1360 }
1361
1362 ret = send_command(sock, RELAYD_CREATE_TRACE_CHUNK, payload.data,
1363 payload.size, 0);
1364 if (ret < 0) {
1365 ERR("Failed to send trace chunk creation command to relay daemon");
1366 goto end;
1367 }
1368
1369 ret = recv_reply(sock, &reply, sizeof(reply));
1370 if (ret < 0) {
1371 ERR("Failed to receive relay daemon trace chunk creation command reply");
1372 goto end;
1373 }
1374
1375 reply.ret_code = be32toh(reply.ret_code);
1376 if (reply.ret_code != LTTNG_OK) {
1377 ret = -1;
1378 ERR("Relayd trace chunk create replied error %d",
1379 reply.ret_code);
1380 } else {
1381 ret = 0;
1382 DBG("Relayd successfully created trace chunk: chunk_id = %" PRIu64,
1383 chunk_id);
1384 }
1385
1386 end:
1387 lttng_dynamic_buffer_reset(&payload);
1388 return ret;
1389 }
1390
1391 int relayd_close_trace_chunk(struct lttcomm_relayd_sock *sock,
1392 struct lttng_trace_chunk *chunk,
1393 char *path)
1394 {
1395 int ret = 0;
1396 enum lttng_trace_chunk_status status;
1397 struct lttcomm_relayd_close_trace_chunk msg = {};
1398 struct lttcomm_relayd_close_trace_chunk_reply reply = {};
1399 uint64_t chunk_id;
1400 time_t close_timestamp;
1401 LTTNG_OPTIONAL(enum lttng_trace_chunk_command_type) close_command = {};
1402
1403 if (!relayd_supports_chunks(sock)) {
1404 DBG("Refusing to close remote trace chunk: relayd does not support chunks");
1405 goto end;
1406 }
1407
1408 status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1409 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1410 ERR("Failed to get trace chunk id");
1411 ret = -1;
1412 goto end;
1413 }
1414
1415 status = lttng_trace_chunk_get_close_timestamp(chunk, &close_timestamp);
1416 if (status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1417 ERR("Failed to get trace chunk close timestamp");
1418 ret = -1;
1419 goto end;
1420 }
1421
1422 status = lttng_trace_chunk_get_close_command(chunk,
1423 &close_command.value);
1424 switch (status) {
1425 case LTTNG_TRACE_CHUNK_STATUS_OK:
1426 close_command.is_set = 1;
1427 break;
1428 case LTTNG_TRACE_CHUNK_STATUS_NONE:
1429 break;
1430 default:
1431 ERR("Failed to get trace chunk close command");
1432 ret = -1;
1433 goto end;
1434 }
1435
1436 msg = (typeof(msg)){
1437 .chunk_id = htobe64(chunk_id),
1438 .close_timestamp = htobe64((uint64_t) close_timestamp),
1439 .close_command = {
1440 .value = htobe32((uint32_t) close_command.value),
1441 .is_set = close_command.is_set,
1442 },
1443 };
1444
1445 ret = send_command(sock, RELAYD_CLOSE_TRACE_CHUNK, &msg, sizeof(msg),
1446 0);
1447 if (ret < 0) {
1448 ERR("Failed to send trace chunk close command to relay daemon");
1449 goto end;
1450 }
1451
1452 ret = recv_reply(sock, &reply, sizeof(reply));
1453 if (ret < 0) {
1454 ERR("Failed to receive relay daemon trace chunk close command reply");
1455 goto end;
1456 }
1457
1458 reply.path_length = be32toh(reply.path_length);
1459 if (reply.path_length >= LTTNG_PATH_MAX) {
1460 ERR("Chunk path too long");
1461 ret = -1;
1462 goto end;
1463 }
1464
1465 ret = recv_reply(sock, path, reply.path_length);
1466 if (ret < 0) {
1467 ERR("Failed to receive relay daemon trace chunk close command reply");
1468 goto end;
1469 }
1470 if (path[reply.path_length - 1] != '\0') {
1471 ERR("Invalid trace chunk path returned by relay daemon (not null-terminated)");
1472 ret = -1;
1473 goto end;
1474 }
1475
1476 reply.generic.ret_code = be32toh(reply.generic.ret_code);
1477 if (reply.generic.ret_code != LTTNG_OK) {
1478 ret = -1;
1479 ERR("Relayd trace chunk close replied error %d",
1480 reply.generic.ret_code);
1481 } else {
1482 ret = 0;
1483 DBG("Relayd successfully closed trace chunk: chunk_id = %" PRIu64,
1484 chunk_id);
1485 }
1486 end:
1487 return ret;
1488 }
1489
1490 int relayd_trace_chunk_exists(struct lttcomm_relayd_sock *sock,
1491 uint64_t chunk_id, bool *chunk_exists)
1492 {
1493 int ret = 0;
1494 struct lttcomm_relayd_trace_chunk_exists msg = {};
1495 struct lttcomm_relayd_trace_chunk_exists_reply reply = {};
1496
1497 if (!relayd_supports_chunks(sock)) {
1498 DBG("Refusing to check for trace chunk existence: relayd does not support chunks");
1499 /* The chunk will never exist */
1500 *chunk_exists = false;
1501 goto end;
1502 }
1503
1504 msg = (typeof(msg)){
1505 .chunk_id = htobe64(chunk_id),
1506 };
1507
1508 ret = send_command(sock, RELAYD_TRACE_CHUNK_EXISTS, &msg, sizeof(msg),
1509 0);
1510 if (ret < 0) {
1511 ERR("Failed to send trace chunk exists command to relay daemon");
1512 goto end;
1513 }
1514
1515 ret = recv_reply(sock, &reply, sizeof(reply));
1516 if (ret < 0) {
1517 ERR("Failed to receive relay daemon trace chunk close command reply");
1518 goto end;
1519 }
1520
1521 reply.generic.ret_code = be32toh(reply.generic.ret_code);
1522 if (reply.generic.ret_code != LTTNG_OK) {
1523 ret = -1;
1524 ERR("Relayd trace chunk close replied error %d",
1525 reply.generic.ret_code);
1526 } else {
1527 ret = 0;
1528 DBG("Relayd successfully checked trace chunk existence: chunk_id = %" PRIu64
1529 ", exists = %s", chunk_id,
1530 reply.trace_chunk_exists ? "true" : "false");
1531 *chunk_exists = !!reply.trace_chunk_exists;
1532 }
1533 end:
1534 return ret;
1535 }
This page took 0.10125 seconds and 3 git commands to generate.