Commit | Line | Data |
---|---|---|
00e2e675 DG |
1 | /* |
2 | * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com> | |
3 | * | |
4 | * This program is free software; you can redistribute it and/or modify it | |
5 | * under the terms of the GNU General Public License, version 2 only, as | |
6 | * published by the Free Software Foundation. | |
7 | * | |
8 | * This program is distributed in the hope that it will be useful, but WITHOUT | |
9 | * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or | |
10 | * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for | |
11 | * more details. | |
12 | * | |
13 | * You should have received a copy of the GNU General Public License along with | |
14 | * this program; if not, write to the Free Software Foundation, Inc., 51 | |
15 | * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. | |
16 | */ | |
17 | ||
6c1c0768 | 18 | #define _LGPL_SOURCE |
00e2e675 DG |
19 | #include <assert.h> |
20 | #include <stdio.h> | |
21 | #include <stdlib.h> | |
22 | #include <string.h> | |
23 | #include <sys/stat.h> | |
77c7c900 | 24 | #include <inttypes.h> |
00e2e675 DG |
25 | |
26 | #include <common/common.h> | |
27 | #include <common/defaults.h> | |
f263b7fd | 28 | #include <common/compat/endian.h> |
00e2e675 | 29 | #include <common/sessiond-comm/relayd.h> |
50adc264 | 30 | #include <common/index/ctf-index.h> |
00e2e675 DG |
31 | |
32 | #include "relayd.h" | |
33 | ||
34 | /* | |
35 | * Send command. Fill up the header and append the data. | |
36 | */ | |
6151a90f | 37 | static int send_command(struct lttcomm_relayd_sock *rsock, |
7c9534d6 | 38 | enum lttcomm_relayd_command cmd, void *data, size_t size, |
00e2e675 DG |
39 | int flags) |
40 | { | |
41 | int ret; | |
42 | struct lttcomm_relayd_hdr header; | |
43 | char *buf; | |
44 | uint64_t buf_size = sizeof(header); | |
45 | ||
f96e4545 MD |
46 | if (rsock->sock.fd < 0) { |
47 | return -ECONNRESET; | |
48 | } | |
49 | ||
00e2e675 DG |
50 | if (data) { |
51 | buf_size += size; | |
52 | } | |
53 | ||
54 | buf = zmalloc(buf_size); | |
55 | if (buf == NULL) { | |
56 | PERROR("zmalloc relayd send command buf"); | |
57 | ret = -1; | |
58 | goto alloc_error; | |
59 | } | |
60 | ||
53efb85a | 61 | memset(&header, 0, sizeof(header)); |
00e2e675 DG |
62 | header.cmd = htobe32(cmd); |
63 | header.data_size = htobe64(size); | |
64 | ||
65 | /* Zeroed for now since not used. */ | |
66 | header.cmd_version = 0; | |
67 | header.circuit_id = 0; | |
68 | ||
69 | /* Prepare buffer to send. */ | |
70 | memcpy(buf, &header, sizeof(header)); | |
71 | if (data) { | |
72 | memcpy(buf + sizeof(header), data, size); | |
73 | } | |
74 | ||
6151a90f | 75 | ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags); |
00e2e675 | 76 | if (ret < 0) { |
8994307f | 77 | ret = -errno; |
00e2e675 DG |
78 | goto error; |
79 | } | |
80 | ||
633d0084 | 81 | DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size); |
00e2e675 DG |
82 | |
83 | error: | |
84 | free(buf); | |
85 | alloc_error: | |
86 | return ret; | |
87 | } | |
88 | ||
89 | /* | |
90 | * Receive reply data on socket. This MUST be call after send_command or else | |
91 | * could result in unexpected behavior(s). | |
92 | */ | |
6151a90f | 93 | static int recv_reply(struct lttcomm_relayd_sock *rsock, void *data, size_t size) |
00e2e675 DG |
94 | { |
95 | int ret; | |
96 | ||
f96e4545 MD |
97 | if (rsock->sock.fd < 0) { |
98 | return -ECONNRESET; | |
99 | } | |
100 | ||
8fd623e0 | 101 | DBG3("Relayd waiting for reply of size %zu", size); |
00e2e675 | 102 | |
6151a90f | 103 | ret = rsock->sock.ops->recvmsg(&rsock->sock, data, size, 0); |
20275fe8 DG |
104 | if (ret <= 0 || ret != size) { |
105 | if (ret == 0) { | |
106 | /* Orderly shutdown. */ | |
6151a90f | 107 | DBG("Socket %d has performed an orderly shutdown", rsock->sock.fd); |
20275fe8 | 108 | } else { |
8fd623e0 | 109 | DBG("Receiving reply failed on sock %d for size %zu with ret %d", |
6151a90f | 110 | rsock->sock.fd, size, ret); |
20275fe8 DG |
111 | } |
112 | /* Always return -1 here and the caller can use errno. */ | |
113 | ret = -1; | |
00e2e675 DG |
114 | goto error; |
115 | } | |
116 | ||
117 | error: | |
118 | return ret; | |
119 | } | |
120 | ||
d3e2ba59 JD |
121 | /* |
122 | * Starting at 2.4, RELAYD_CREATE_SESSION takes additional parameters to | |
123 | * support the live reading capability. | |
124 | */ | |
125 | static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock, | |
126 | uint64_t *session_id, char *session_name, char *hostname, | |
7d2f7452 | 127 | int session_live_timer, unsigned int snapshot) |
d3e2ba59 JD |
128 | { |
129 | int ret; | |
130 | struct lttcomm_relayd_create_session_2_4 msg; | |
131 | ||
3a13ffd5 MD |
132 | if (lttng_strncpy(msg.session_name, session_name, |
133 | sizeof(msg.session_name))) { | |
246777db MD |
134 | ret = -1; |
135 | goto error; | |
136 | } | |
3a13ffd5 | 137 | if (lttng_strncpy(msg.hostname, hostname, sizeof(msg.hostname))) { |
246777db MD |
138 | ret = -1; |
139 | goto error; | |
140 | } | |
d3e2ba59 | 141 | msg.live_timer = htobe32(session_live_timer); |
7d2f7452 | 142 | msg.snapshot = htobe32(snapshot); |
d3e2ba59 JD |
143 | |
144 | /* Send command */ | |
145 | ret = send_command(rsock, RELAYD_CREATE_SESSION, &msg, sizeof(msg), 0); | |
146 | if (ret < 0) { | |
147 | goto error; | |
148 | } | |
149 | ||
150 | error: | |
151 | return ret; | |
152 | } | |
153 | ||
154 | /* | |
155 | * RELAYD_CREATE_SESSION from 2.1 to 2.3. | |
156 | */ | |
157 | static int relayd_create_session_2_1(struct lttcomm_relayd_sock *rsock, | |
158 | uint64_t *session_id) | |
159 | { | |
160 | int ret; | |
161 | ||
162 | /* Send command */ | |
163 | ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0); | |
164 | if (ret < 0) { | |
165 | goto error; | |
166 | } | |
167 | ||
168 | error: | |
169 | return ret; | |
170 | } | |
171 | ||
c5b6f4f0 DG |
172 | /* |
173 | * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and | |
174 | * set session_id of the relayd if we have a successful reply from the relayd. | |
175 | * | |
20275fe8 DG |
176 | * On success, return 0 else a negative value which is either an errno error or |
177 | * a lttng error code from the relayd. | |
c5b6f4f0 | 178 | */ |
d3e2ba59 | 179 | int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_id, |
7d2f7452 DG |
180 | char *session_name, char *hostname, int session_live_timer, |
181 | unsigned int snapshot) | |
c5b6f4f0 DG |
182 | { |
183 | int ret; | |
184 | struct lttcomm_relayd_status_session reply; | |
185 | ||
6151a90f | 186 | assert(rsock); |
c5b6f4f0 DG |
187 | assert(session_id); |
188 | ||
189 | DBG("Relayd create session"); | |
190 | ||
d3e2ba59 JD |
191 | switch(rsock->minor) { |
192 | case 1: | |
193 | case 2: | |
194 | case 3: | |
195 | ret = relayd_create_session_2_1(rsock, session_id); | |
31d74dc3 | 196 | break; |
d3e2ba59 JD |
197 | case 4: |
198 | default: | |
7d2f7452 DG |
199 | ret = relayd_create_session_2_4(rsock, session_id, session_name, |
200 | hostname, session_live_timer, snapshot); | |
31d74dc3 | 201 | break; |
d3e2ba59 JD |
202 | } |
203 | ||
c5b6f4f0 DG |
204 | if (ret < 0) { |
205 | goto error; | |
206 | } | |
207 | ||
20275fe8 | 208 | /* Receive response */ |
6151a90f | 209 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
c5b6f4f0 DG |
210 | if (ret < 0) { |
211 | goto error; | |
212 | } | |
213 | ||
214 | reply.session_id = be64toh(reply.session_id); | |
215 | reply.ret_code = be32toh(reply.ret_code); | |
216 | ||
217 | /* Return session id or negative ret code. */ | |
218 | if (reply.ret_code != LTTNG_OK) { | |
bb63afd9 DG |
219 | ret = -1; |
220 | ERR("Relayd create session replied error %d", reply.ret_code); | |
c5b6f4f0 DG |
221 | goto error; |
222 | } else { | |
223 | ret = 0; | |
224 | *session_id = reply.session_id; | |
225 | } | |
226 | ||
227 | DBG("Relayd session created with id %" PRIu64, reply.session_id); | |
228 | ||
229 | error: | |
230 | return ret; | |
231 | } | |
232 | ||
00e2e675 DG |
233 | /* |
234 | * Add stream on the relayd and assign stream handle to the stream_id argument. | |
235 | * | |
236 | * On success return 0 else return ret_code negative value. | |
237 | */ | |
6151a90f | 238 | int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name, |
0f907de1 | 239 | const char *pathname, uint64_t *stream_id, |
ccb5d350 JD |
240 | uint64_t tracefile_size, uint64_t tracefile_count, |
241 | enum lttng_domain_type domain) | |
00e2e675 DG |
242 | { |
243 | int ret; | |
244 | struct lttcomm_relayd_add_stream msg; | |
0f907de1 | 245 | struct lttcomm_relayd_add_stream_2_2 msg_2_2; |
ccb5d350 | 246 | struct lttcomm_relayd_add_stream_2_11 msg_2_11; |
00e2e675 DG |
247 | struct lttcomm_relayd_status_stream reply; |
248 | ||
249 | /* Code flow error. Safety net. */ | |
6151a90f | 250 | assert(rsock); |
00e2e675 DG |
251 | assert(channel_name); |
252 | assert(pathname); | |
253 | ||
254 | DBG("Relayd adding stream for channel name %s", channel_name); | |
255 | ||
0f907de1 | 256 | /* Compat with relayd 2.1 */ |
ccb5d350 JD |
257 | switch (rsock->minor) { |
258 | case 1: | |
53efb85a | 259 | memset(&msg, 0, sizeof(msg)); |
a7c918ad MD |
260 | if (lttng_strncpy(msg.channel_name, channel_name, |
261 | sizeof(msg.channel_name))) { | |
246777db MD |
262 | ret = -1; |
263 | goto error; | |
264 | } | |
a7c918ad MD |
265 | if (lttng_strncpy(msg.pathname, pathname, |
266 | sizeof(msg.pathname))) { | |
246777db MD |
267 | ret = -1; |
268 | goto error; | |
269 | } | |
00e2e675 | 270 | |
0f907de1 JD |
271 | /* Send command */ |
272 | ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0); | |
273 | if (ret < 0) { | |
274 | goto error; | |
275 | } | |
ccb5d350 JD |
276 | break; |
277 | case 2: | |
278 | case 3: | |
279 | case 4: | |
280 | case 5: | |
281 | case 6: | |
282 | case 7: | |
283 | case 8: | |
284 | case 9: | |
285 | case 10: | |
53efb85a | 286 | memset(&msg_2_2, 0, sizeof(msg_2_2)); |
0f907de1 | 287 | /* Compat with relayd 2.2+ */ |
a7c918ad MD |
288 | if (lttng_strncpy(msg_2_2.channel_name, channel_name, |
289 | sizeof(msg_2_2.channel_name))) { | |
246777db MD |
290 | ret = -1; |
291 | goto error; | |
292 | } | |
a7c918ad MD |
293 | if (lttng_strncpy(msg_2_2.pathname, pathname, |
294 | sizeof(msg_2_2.pathname))) { | |
246777db MD |
295 | ret = -1; |
296 | goto error; | |
297 | } | |
0f907de1 JD |
298 | msg_2_2.tracefile_size = htobe64(tracefile_size); |
299 | msg_2_2.tracefile_count = htobe64(tracefile_count); | |
300 | ||
301 | /* Send command */ | |
302 | ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg_2_2, sizeof(msg_2_2), 0); | |
303 | if (ret < 0) { | |
304 | goto error; | |
305 | } | |
ccb5d350 JD |
306 | break; |
307 | case 11: | |
308 | default: | |
309 | memset(&msg_2_11, 0, sizeof(msg_2_11)); | |
310 | /* Compat with relayd 2.11+ */ | |
311 | if (lttng_strncpy(msg_2_11.channel_name, channel_name, | |
312 | sizeof(msg_2_11.channel_name))) { | |
313 | ret = -1; | |
314 | goto error; | |
315 | } | |
316 | if (lttng_strncpy(msg_2_11.pathname, pathname, | |
317 | sizeof(msg_2_11.pathname))) { | |
318 | ret = -1; | |
319 | goto error; | |
320 | } | |
321 | msg_2_11.tracefile_size = htobe64(tracefile_size); | |
322 | msg_2_11.tracefile_count = htobe64(tracefile_count); | |
323 | msg_2_11.domain = htobe32(domain); | |
324 | ||
325 | /* Send command */ | |
326 | ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg_2_11, sizeof(msg_2_11), 0); | |
327 | if (ret < 0) { | |
328 | goto error; | |
329 | } | |
330 | break; | |
00e2e675 DG |
331 | } |
332 | ||
633d0084 | 333 | /* Waiting for reply */ |
6151a90f | 334 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
00e2e675 DG |
335 | if (ret < 0) { |
336 | goto error; | |
337 | } | |
338 | ||
339 | /* Back to host bytes order. */ | |
340 | reply.handle = be64toh(reply.handle); | |
341 | reply.ret_code = be32toh(reply.ret_code); | |
342 | ||
343 | /* Return session id or negative ret code. */ | |
f73fabfd | 344 | if (reply.ret_code != LTTNG_OK) { |
bb63afd9 DG |
345 | ret = -1; |
346 | ERR("Relayd add stream replied error %d", reply.ret_code); | |
00e2e675 DG |
347 | } else { |
348 | /* Success */ | |
349 | ret = 0; | |
350 | *stream_id = reply.handle; | |
351 | } | |
352 | ||
77c7c900 MD |
353 | DBG("Relayd stream added successfully with handle %" PRIu64, |
354 | reply.handle); | |
00e2e675 DG |
355 | |
356 | error: | |
357 | return ret; | |
358 | } | |
359 | ||
a4baae1b JD |
360 | /* |
361 | * Inform the relay that all the streams for the current channel has been sent. | |
362 | * | |
363 | * On success return 0 else return ret_code negative value. | |
364 | */ | |
365 | int relayd_streams_sent(struct lttcomm_relayd_sock *rsock) | |
366 | { | |
367 | int ret; | |
368 | struct lttcomm_relayd_generic_reply reply; | |
369 | ||
370 | /* Code flow error. Safety net. */ | |
371 | assert(rsock); | |
372 | ||
373 | DBG("Relayd sending streams sent."); | |
374 | ||
375 | /* This feature was introduced in 2.4, ignore it for earlier versions. */ | |
376 | if (rsock->minor < 4) { | |
377 | ret = 0; | |
378 | goto end; | |
379 | } | |
380 | ||
381 | /* Send command */ | |
382 | ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0); | |
383 | if (ret < 0) { | |
384 | goto error; | |
385 | } | |
386 | ||
387 | /* Waiting for reply */ | |
388 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); | |
389 | if (ret < 0) { | |
390 | goto error; | |
391 | } | |
392 | ||
393 | /* Back to host bytes order. */ | |
394 | reply.ret_code = be32toh(reply.ret_code); | |
395 | ||
396 | /* Return session id or negative ret code. */ | |
397 | if (reply.ret_code != LTTNG_OK) { | |
398 | ret = -1; | |
399 | ERR("Relayd streams sent replied error %d", reply.ret_code); | |
400 | goto error; | |
401 | } else { | |
402 | /* Success */ | |
403 | ret = 0; | |
404 | } | |
405 | ||
406 | DBG("Relayd streams sent success"); | |
407 | ||
408 | error: | |
409 | end: | |
410 | return ret; | |
411 | } | |
412 | ||
00e2e675 DG |
413 | /* |
414 | * Check version numbers on the relayd. | |
d4519fa3 JD |
415 | * If major versions are compatible, we assign minor_to_use to the |
416 | * minor version of the procotol we are going to use for this session. | |
00e2e675 DG |
417 | * |
418 | * Return 0 if compatible else negative value. | |
419 | */ | |
6151a90f | 420 | int relayd_version_check(struct lttcomm_relayd_sock *rsock) |
00e2e675 DG |
421 | { |
422 | int ret; | |
092b6259 | 423 | struct lttcomm_relayd_version msg; |
00e2e675 DG |
424 | |
425 | /* Code flow error. Safety net. */ | |
6151a90f | 426 | assert(rsock); |
00e2e675 | 427 | |
6151a90f JD |
428 | DBG("Relayd version check for major.minor %u.%u", rsock->major, |
429 | rsock->minor); | |
00e2e675 | 430 | |
53efb85a | 431 | memset(&msg, 0, sizeof(msg)); |
092b6259 | 432 | /* Prepare network byte order before transmission. */ |
6151a90f JD |
433 | msg.major = htobe32(rsock->major); |
434 | msg.minor = htobe32(rsock->minor); | |
092b6259 | 435 | |
00e2e675 | 436 | /* Send command */ |
6151a90f | 437 | ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0); |
00e2e675 DG |
438 | if (ret < 0) { |
439 | goto error; | |
440 | } | |
441 | ||
20275fe8 | 442 | /* Receive response */ |
6151a90f | 443 | ret = recv_reply(rsock, (void *) &msg, sizeof(msg)); |
00e2e675 DG |
444 | if (ret < 0) { |
445 | goto error; | |
446 | } | |
447 | ||
448 | /* Set back to host bytes order */ | |
092b6259 DG |
449 | msg.major = be32toh(msg.major); |
450 | msg.minor = be32toh(msg.minor); | |
451 | ||
452 | /* | |
453 | * Only validate the major version. If the other side is higher, | |
454 | * communication is not possible. Only major version equal can talk to each | |
455 | * other. If the minor version differs, the lowest version is used by both | |
456 | * sides. | |
092b6259 | 457 | */ |
6151a90f | 458 | if (msg.major != rsock->major) { |
d4519fa3 JD |
459 | /* Not compatible */ |
460 | ret = -1; | |
461 | DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)", | |
6151a90f | 462 | msg.major, rsock->major); |
092b6259 | 463 | goto error; |
00e2e675 DG |
464 | } |
465 | ||
092b6259 | 466 | /* |
6151a90f JD |
467 | * If the relayd's minor version is higher, it will adapt to our version so |
468 | * we can continue to use the latest relayd communication data structure. | |
469 | * If the received minor version is higher, the relayd should adapt to us. | |
092b6259 | 470 | */ |
6151a90f JD |
471 | if (rsock->minor > msg.minor) { |
472 | rsock->minor = msg.minor; | |
d4519fa3 | 473 | } |
092b6259 | 474 | |
d4519fa3 JD |
475 | /* Version number compatible */ |
476 | DBG2("Relayd version is compatible, using protocol version %u.%u", | |
6151a90f | 477 | rsock->major, rsock->minor); |
d4519fa3 | 478 | ret = 0; |
00e2e675 DG |
479 | |
480 | error: | |
481 | return ret; | |
482 | } | |
483 | ||
00e2e675 DG |
484 | /* |
485 | * Add stream on the relayd and assign stream handle to the stream_id argument. | |
486 | * | |
487 | * On success return 0 else return ret_code negative value. | |
488 | */ | |
6151a90f | 489 | int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len) |
00e2e675 DG |
490 | { |
491 | int ret; | |
492 | ||
493 | /* Code flow error. Safety net. */ | |
6151a90f | 494 | assert(rsock); |
00e2e675 | 495 | |
77c7c900 | 496 | DBG("Relayd sending metadata of size %zu", len); |
00e2e675 DG |
497 | |
498 | /* Send command */ | |
6151a90f | 499 | ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0); |
00e2e675 DG |
500 | if (ret < 0) { |
501 | goto error; | |
502 | } | |
503 | ||
504 | DBG2("Relayd metadata added successfully"); | |
505 | ||
506 | /* | |
507 | * After that call, the metadata data MUST be sent to the relayd so the | |
508 | * receive size on the other end matches the len of the metadata packet | |
633d0084 | 509 | * header. This is why we don't wait for a reply here. |
00e2e675 DG |
510 | */ |
511 | ||
512 | error: | |
513 | return ret; | |
514 | } | |
515 | ||
516 | /* | |
6151a90f | 517 | * Connect to relay daemon with an allocated lttcomm_relayd_sock. |
00e2e675 | 518 | */ |
6151a90f | 519 | int relayd_connect(struct lttcomm_relayd_sock *rsock) |
00e2e675 DG |
520 | { |
521 | /* Code flow error. Safety net. */ | |
6151a90f | 522 | assert(rsock); |
00e2e675 | 523 | |
f96e4545 MD |
524 | if (!rsock->sock.ops) { |
525 | /* | |
526 | * Attempting a connect on a non-initialized socket. | |
527 | */ | |
528 | return -ECONNRESET; | |
529 | } | |
530 | ||
00e2e675 DG |
531 | DBG3("Relayd connect ..."); |
532 | ||
6151a90f | 533 | return rsock->sock.ops->connect(&rsock->sock); |
00e2e675 DG |
534 | } |
535 | ||
536 | /* | |
6151a90f | 537 | * Close relayd socket with an allocated lttcomm_relayd_sock. |
ffe60014 DG |
538 | * |
539 | * If no socket operations are found, simply return 0 meaning that everything | |
540 | * is fine. Without operations, the socket can not possibly be opened or used. | |
541 | * This is possible if the socket was allocated but not created. However, the | |
542 | * caller could simply use it to store a valid file descriptor for instance | |
543 | * passed over a Unix socket and call this to cleanup but still without a valid | |
544 | * ops pointer. | |
545 | * | |
546 | * Return the close returned value. On error, a negative value is usually | |
547 | * returned back from close(2). | |
00e2e675 | 548 | */ |
6151a90f | 549 | int relayd_close(struct lttcomm_relayd_sock *rsock) |
00e2e675 | 550 | { |
ffe60014 DG |
551 | int ret; |
552 | ||
00e2e675 | 553 | /* Code flow error. Safety net. */ |
6151a90f | 554 | assert(rsock); |
00e2e675 | 555 | |
ffe60014 | 556 | /* An invalid fd is fine, return success. */ |
6151a90f | 557 | if (rsock->sock.fd < 0) { |
ffe60014 DG |
558 | ret = 0; |
559 | goto end; | |
560 | } | |
561 | ||
6151a90f | 562 | DBG3("Relayd closing socket %d", rsock->sock.fd); |
00e2e675 | 563 | |
6151a90f JD |
564 | if (rsock->sock.ops) { |
565 | ret = rsock->sock.ops->close(&rsock->sock); | |
ffe60014 DG |
566 | } else { |
567 | /* Default call if no specific ops found. */ | |
6151a90f | 568 | ret = close(rsock->sock.fd); |
ffe60014 DG |
569 | if (ret < 0) { |
570 | PERROR("relayd_close default close"); | |
571 | } | |
572 | } | |
f96e4545 | 573 | rsock->sock.fd = -1; |
ffe60014 DG |
574 | |
575 | end: | |
576 | return ret; | |
00e2e675 DG |
577 | } |
578 | ||
579 | /* | |
580 | * Send data header structure to the relayd. | |
581 | */ | |
6151a90f | 582 | int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock, |
00e2e675 DG |
583 | struct lttcomm_relayd_data_hdr *hdr, size_t size) |
584 | { | |
585 | int ret; | |
586 | ||
587 | /* Code flow error. Safety net. */ | |
6151a90f | 588 | assert(rsock); |
00e2e675 DG |
589 | assert(hdr); |
590 | ||
f96e4545 MD |
591 | if (rsock->sock.fd < 0) { |
592 | return -ECONNRESET; | |
593 | } | |
594 | ||
8fd623e0 | 595 | DBG3("Relayd sending data header of size %zu", size); |
00e2e675 DG |
596 | |
597 | /* Again, safety net */ | |
598 | if (size == 0) { | |
599 | size = sizeof(struct lttcomm_relayd_data_hdr); | |
600 | } | |
601 | ||
602 | /* Only send data header. */ | |
6151a90f | 603 | ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0); |
00e2e675 | 604 | if (ret < 0) { |
8994307f | 605 | ret = -errno; |
00e2e675 DG |
606 | goto error; |
607 | } | |
608 | ||
609 | /* | |
610 | * The data MUST be sent right after that command for the receive on the | |
611 | * other end to match the size in the header. | |
612 | */ | |
613 | ||
614 | error: | |
615 | return ret; | |
616 | } | |
173af62f DG |
617 | |
618 | /* | |
619 | * Send close stream command to the relayd. | |
620 | */ | |
6151a90f | 621 | int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, |
173af62f DG |
622 | uint64_t last_net_seq_num) |
623 | { | |
624 | int ret; | |
625 | struct lttcomm_relayd_close_stream msg; | |
626 | struct lttcomm_relayd_generic_reply reply; | |
627 | ||
628 | /* Code flow error. Safety net. */ | |
6151a90f | 629 | assert(rsock); |
173af62f | 630 | |
77c7c900 | 631 | DBG("Relayd closing stream id %" PRIu64, stream_id); |
173af62f | 632 | |
53efb85a | 633 | memset(&msg, 0, sizeof(msg)); |
173af62f DG |
634 | msg.stream_id = htobe64(stream_id); |
635 | msg.last_net_seq_num = htobe64(last_net_seq_num); | |
636 | ||
637 | /* Send command */ | |
6151a90f | 638 | ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0); |
173af62f DG |
639 | if (ret < 0) { |
640 | goto error; | |
641 | } | |
642 | ||
20275fe8 | 643 | /* Receive response */ |
6151a90f | 644 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
173af62f DG |
645 | if (ret < 0) { |
646 | goto error; | |
647 | } | |
648 | ||
649 | reply.ret_code = be32toh(reply.ret_code); | |
650 | ||
651 | /* Return session id or negative ret code. */ | |
f73fabfd | 652 | if (reply.ret_code != LTTNG_OK) { |
bb63afd9 DG |
653 | ret = -1; |
654 | ERR("Relayd close stream replied error %d", reply.ret_code); | |
173af62f DG |
655 | } else { |
656 | /* Success */ | |
657 | ret = 0; | |
658 | } | |
659 | ||
77c7c900 | 660 | DBG("Relayd close stream id %" PRIu64 " successfully", stream_id); |
173af62f DG |
661 | |
662 | error: | |
663 | return ret; | |
664 | } | |
c8f59ee5 DG |
665 | |
666 | /* | |
667 | * Check for data availability for a given stream id. | |
668 | * | |
6d805429 | 669 | * Return 0 if NOT pending, 1 if so and a negative value on error. |
c8f59ee5 | 670 | */ |
6151a90f | 671 | int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, |
c8f59ee5 DG |
672 | uint64_t last_net_seq_num) |
673 | { | |
674 | int ret; | |
6d805429 | 675 | struct lttcomm_relayd_data_pending msg; |
c8f59ee5 DG |
676 | struct lttcomm_relayd_generic_reply reply; |
677 | ||
678 | /* Code flow error. Safety net. */ | |
6151a90f | 679 | assert(rsock); |
c8f59ee5 | 680 | |
6d805429 | 681 | DBG("Relayd data pending for stream id %" PRIu64, stream_id); |
c8f59ee5 | 682 | |
53efb85a | 683 | memset(&msg, 0, sizeof(msg)); |
c8f59ee5 DG |
684 | msg.stream_id = htobe64(stream_id); |
685 | msg.last_net_seq_num = htobe64(last_net_seq_num); | |
686 | ||
687 | /* Send command */ | |
6151a90f | 688 | ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg, |
c8f59ee5 DG |
689 | sizeof(msg), 0); |
690 | if (ret < 0) { | |
691 | goto error; | |
692 | } | |
693 | ||
20275fe8 | 694 | /* Receive response */ |
6151a90f | 695 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
c8f59ee5 DG |
696 | if (ret < 0) { |
697 | goto error; | |
698 | } | |
699 | ||
700 | reply.ret_code = be32toh(reply.ret_code); | |
701 | ||
702 | /* Return session id or negative ret code. */ | |
703 | if (reply.ret_code >= LTTNG_OK) { | |
bb63afd9 | 704 | ERR("Relayd data pending replied error %d", reply.ret_code); |
c8f59ee5 DG |
705 | } |
706 | ||
707 | /* At this point, the ret code is either 1 or 0 */ | |
708 | ret = reply.ret_code; | |
709 | ||
6d805429 | 710 | DBG("Relayd data is %s pending for stream id %" PRIu64, |
9dd26bb9 | 711 | ret == 1 ? "" : "NOT", stream_id); |
c8f59ee5 DG |
712 | |
713 | error: | |
714 | return ret; | |
715 | } | |
716 | ||
717 | /* | |
718 | * Check on the relayd side for a quiescent state on the control socket. | |
719 | */ | |
6151a90f | 720 | int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock, |
ad7051c0 | 721 | uint64_t metadata_stream_id) |
c8f59ee5 DG |
722 | { |
723 | int ret; | |
ad7051c0 | 724 | struct lttcomm_relayd_quiescent_control msg; |
c8f59ee5 DG |
725 | struct lttcomm_relayd_generic_reply reply; |
726 | ||
727 | /* Code flow error. Safety net. */ | |
6151a90f | 728 | assert(rsock); |
c8f59ee5 DG |
729 | |
730 | DBG("Relayd checking quiescent control state"); | |
731 | ||
53efb85a | 732 | memset(&msg, 0, sizeof(msg)); |
ad7051c0 DG |
733 | msg.stream_id = htobe64(metadata_stream_id); |
734 | ||
c8f59ee5 | 735 | /* Send command */ |
6151a90f | 736 | ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0); |
c8f59ee5 DG |
737 | if (ret < 0) { |
738 | goto error; | |
739 | } | |
740 | ||
20275fe8 | 741 | /* Receive response */ |
6151a90f | 742 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
c8f59ee5 DG |
743 | if (ret < 0) { |
744 | goto error; | |
745 | } | |
746 | ||
747 | reply.ret_code = be32toh(reply.ret_code); | |
748 | ||
749 | /* Return session id or negative ret code. */ | |
750 | if (reply.ret_code != LTTNG_OK) { | |
bb63afd9 DG |
751 | ret = -1; |
752 | ERR("Relayd quiescent control replied error %d", reply.ret_code); | |
c8f59ee5 DG |
753 | goto error; |
754 | } | |
755 | ||
756 | /* Control socket is quiescent */ | |
6d805429 | 757 | return 0; |
c8f59ee5 DG |
758 | |
759 | error: | |
760 | return ret; | |
761 | } | |
f7079f67 DG |
762 | |
763 | /* | |
764 | * Begin a data pending command for a specific session id. | |
765 | */ | |
6151a90f | 766 | int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id) |
f7079f67 DG |
767 | { |
768 | int ret; | |
769 | struct lttcomm_relayd_begin_data_pending msg; | |
770 | struct lttcomm_relayd_generic_reply reply; | |
771 | ||
772 | /* Code flow error. Safety net. */ | |
6151a90f | 773 | assert(rsock); |
f7079f67 DG |
774 | |
775 | DBG("Relayd begin data pending"); | |
776 | ||
53efb85a | 777 | memset(&msg, 0, sizeof(msg)); |
f7079f67 DG |
778 | msg.session_id = htobe64(id); |
779 | ||
780 | /* Send command */ | |
6151a90f | 781 | ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0); |
f7079f67 DG |
782 | if (ret < 0) { |
783 | goto error; | |
784 | } | |
785 | ||
20275fe8 | 786 | /* Receive response */ |
6151a90f | 787 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
f7079f67 DG |
788 | if (ret < 0) { |
789 | goto error; | |
790 | } | |
791 | ||
792 | reply.ret_code = be32toh(reply.ret_code); | |
793 | ||
794 | /* Return session id or negative ret code. */ | |
795 | if (reply.ret_code != LTTNG_OK) { | |
bb63afd9 DG |
796 | ret = -1; |
797 | ERR("Relayd begin data pending replied error %d", reply.ret_code); | |
f7079f67 DG |
798 | goto error; |
799 | } | |
800 | ||
801 | return 0; | |
802 | ||
803 | error: | |
804 | return ret; | |
805 | } | |
806 | ||
807 | /* | |
808 | * End a data pending command for a specific session id. | |
809 | * | |
810 | * Return 0 on success and set is_data_inflight to 0 if no data is being | |
811 | * streamed or 1 if it is the case. | |
812 | */ | |
6151a90f | 813 | int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id, |
f7079f67 DG |
814 | unsigned int *is_data_inflight) |
815 | { | |
af6c30b5 | 816 | int ret, recv_ret; |
f7079f67 DG |
817 | struct lttcomm_relayd_end_data_pending msg; |
818 | struct lttcomm_relayd_generic_reply reply; | |
819 | ||
820 | /* Code flow error. Safety net. */ | |
6151a90f | 821 | assert(rsock); |
f7079f67 DG |
822 | |
823 | DBG("Relayd end data pending"); | |
824 | ||
53efb85a | 825 | memset(&msg, 0, sizeof(msg)); |
f7079f67 DG |
826 | msg.session_id = htobe64(id); |
827 | ||
828 | /* Send command */ | |
6151a90f | 829 | ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0); |
f7079f67 DG |
830 | if (ret < 0) { |
831 | goto error; | |
832 | } | |
833 | ||
20275fe8 | 834 | /* Receive response */ |
6151a90f | 835 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
f7079f67 DG |
836 | if (ret < 0) { |
837 | goto error; | |
838 | } | |
839 | ||
af6c30b5 DG |
840 | recv_ret = be32toh(reply.ret_code); |
841 | if (recv_ret < 0) { | |
842 | ret = recv_ret; | |
f7079f67 DG |
843 | goto error; |
844 | } | |
845 | ||
af6c30b5 | 846 | *is_data_inflight = recv_ret; |
f7079f67 | 847 | |
af6c30b5 | 848 | DBG("Relayd end data pending is data inflight: %d", recv_ret); |
f7079f67 DG |
849 | |
850 | return 0; | |
851 | ||
852 | error: | |
853 | return ret; | |
854 | } | |
1c20f0e2 JD |
855 | |
856 | /* | |
857 | * Send index to the relayd. | |
858 | */ | |
859 | int relayd_send_index(struct lttcomm_relayd_sock *rsock, | |
50adc264 | 860 | struct ctf_packet_index *index, uint64_t relay_stream_id, |
1c20f0e2 JD |
861 | uint64_t net_seq_num) |
862 | { | |
863 | int ret; | |
864 | struct lttcomm_relayd_index msg; | |
865 | struct lttcomm_relayd_generic_reply reply; | |
866 | ||
867 | /* Code flow error. Safety net. */ | |
868 | assert(rsock); | |
869 | ||
870 | if (rsock->minor < 4) { | |
871 | DBG("Not sending indexes before protocol 2.4"); | |
872 | ret = 0; | |
873 | goto error; | |
874 | } | |
875 | ||
876 | DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id); | |
877 | ||
53efb85a | 878 | memset(&msg, 0, sizeof(msg)); |
1c20f0e2 JD |
879 | msg.relay_stream_id = htobe64(relay_stream_id); |
880 | msg.net_seq_num = htobe64(net_seq_num); | |
881 | ||
882 | /* The index is already in big endian. */ | |
883 | msg.packet_size = index->packet_size; | |
884 | msg.content_size = index->content_size; | |
885 | msg.timestamp_begin = index->timestamp_begin; | |
886 | msg.timestamp_end = index->timestamp_end; | |
887 | msg.events_discarded = index->events_discarded; | |
888 | msg.stream_id = index->stream_id; | |
889 | ||
234cd636 JD |
890 | if (rsock->minor >= 8) { |
891 | msg.stream_instance_id = index->stream_instance_id; | |
892 | msg.packet_seq_num = index->packet_seq_num; | |
893 | } | |
894 | ||
1c20f0e2 | 895 | /* Send command */ |
f8f3885c MD |
896 | ret = send_command(rsock, RELAYD_SEND_INDEX, &msg, |
897 | lttcomm_relayd_index_len(lttng_to_index_major(rsock->major, | |
898 | rsock->minor), | |
899 | lttng_to_index_minor(rsock->major, rsock->minor)), | |
900 | 0); | |
1c20f0e2 JD |
901 | if (ret < 0) { |
902 | goto error; | |
903 | } | |
904 | ||
905 | /* Receive response */ | |
906 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); | |
907 | if (ret < 0) { | |
908 | goto error; | |
909 | } | |
910 | ||
911 | reply.ret_code = be32toh(reply.ret_code); | |
912 | ||
913 | /* Return session id or negative ret code. */ | |
914 | if (reply.ret_code != LTTNG_OK) { | |
915 | ret = -1; | |
916 | ERR("Relayd send index replied error %d", reply.ret_code); | |
917 | } else { | |
918 | /* Success */ | |
919 | ret = 0; | |
920 | } | |
921 | ||
922 | error: | |
923 | return ret; | |
924 | } | |
93ec662e JD |
925 | |
926 | /* | |
927 | * Ask the relay to reset the metadata trace file (regeneration). | |
928 | */ | |
929 | int relayd_reset_metadata(struct lttcomm_relayd_sock *rsock, | |
930 | uint64_t stream_id, uint64_t version) | |
931 | { | |
932 | int ret; | |
933 | struct lttcomm_relayd_reset_metadata msg; | |
934 | struct lttcomm_relayd_generic_reply reply; | |
935 | ||
936 | /* Code flow error. Safety net. */ | |
937 | assert(rsock); | |
938 | ||
939 | /* Should have been prevented by the sessiond. */ | |
940 | if (rsock->minor < 8) { | |
941 | ERR("Metadata regeneration unsupported before 2.8"); | |
942 | ret = -1; | |
943 | goto error; | |
944 | } | |
945 | ||
946 | DBG("Relayd reset metadata stream id %" PRIu64, stream_id); | |
947 | ||
948 | memset(&msg, 0, sizeof(msg)); | |
949 | msg.stream_id = htobe64(stream_id); | |
950 | msg.version = htobe64(version); | |
951 | ||
952 | /* Send command */ | |
953 | ret = send_command(rsock, RELAYD_RESET_METADATA, (void *) &msg, sizeof(msg), 0); | |
954 | if (ret < 0) { | |
955 | goto error; | |
956 | } | |
957 | ||
958 | /* Receive response */ | |
959 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); | |
960 | if (ret < 0) { | |
961 | goto error; | |
962 | } | |
963 | ||
964 | reply.ret_code = be32toh(reply.ret_code); | |
965 | ||
966 | /* Return session id or negative ret code. */ | |
967 | if (reply.ret_code != LTTNG_OK) { | |
968 | ret = -1; | |
969 | ERR("Relayd reset metadata replied error %d", reply.ret_code); | |
970 | } else { | |
971 | /* Success */ | |
972 | ret = 0; | |
973 | } | |
974 | ||
975 | DBG("Relayd reset metadata stream id %" PRIu64 " successfully", stream_id); | |
976 | ||
977 | error: | |
978 | return ret; | |
979 | } | |
d26a0e48 JD |
980 | |
981 | int relayd_rotate_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, | |
2d7d73bc JD |
982 | const char *new_pathname, uint64_t new_chunk_id, |
983 | uint64_t seq_num) | |
d26a0e48 JD |
984 | { |
985 | int ret; | |
986 | struct lttcomm_relayd_rotate_stream msg; | |
987 | struct lttcomm_relayd_generic_reply reply; | |
988 | ||
989 | /* Code flow error. Safety net. */ | |
990 | assert(rsock); | |
991 | ||
992 | DBG("Relayd rotating stream id %" PRIu64, stream_id); | |
993 | ||
994 | memset(&msg, 0, sizeof(msg)); | |
995 | msg.stream_id = htobe64(stream_id); | |
2d7d73bc JD |
996 | msg.rotate_at_seq_num = htobe64(seq_num); |
997 | msg.new_chunk_id = htobe64(new_chunk_id); | |
d26a0e48 JD |
998 | if (lttng_strncpy(msg.new_pathname, new_pathname, |
999 | sizeof(msg.new_pathname))) { | |
1000 | ret = -1; | |
1001 | goto error; | |
1002 | } | |
1003 | ||
1004 | /* Send command */ | |
1005 | ret = send_command(rsock, RELAYD_ROTATE_STREAM, (void *) &msg, sizeof(msg), 0); | |
1006 | if (ret < 0) { | |
1007 | goto error; | |
1008 | } | |
1009 | ||
1010 | /* Receive response */ | |
1011 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); | |
1012 | if (ret < 0) { | |
1013 | goto error; | |
1014 | } | |
1015 | ||
1016 | reply.ret_code = be32toh(reply.ret_code); | |
1017 | ||
1018 | /* Return session id or negative ret code. */ | |
1019 | if (reply.ret_code != LTTNG_OK) { | |
1020 | ret = -1; | |
1021 | ERR("Relayd rotate stream replied error %d", reply.ret_code); | |
1022 | } else { | |
1023 | /* Success */ | |
1024 | ret = 0; | |
1025 | } | |
1026 | ||
1027 | DBG("Relayd rotated stream id %" PRIu64 " successfully", stream_id); | |
1028 | ||
0ee5eef6 JD |
1029 | error: |
1030 | return ret; | |
1031 | } | |
1032 | ||
1033 | int relayd_rotate_rename(struct lttcomm_relayd_sock *rsock, | |
1034 | const char *current_path, const char *new_path) | |
1035 | { | |
1036 | int ret; | |
1037 | struct lttcomm_relayd_rotate_rename msg; | |
1038 | struct lttcomm_relayd_generic_reply reply; | |
1039 | ||
1040 | /* Code flow error. Safety net. */ | |
1041 | assert(rsock); | |
1042 | ||
1043 | DBG("Relayd rename chunk %s to %s", current_path, new_path); | |
1044 | ||
1045 | memset(&msg, 0, sizeof(msg)); | |
1046 | if (lttng_strncpy(msg.current_path, current_path, | |
1047 | sizeof(msg.current_path))) { | |
1048 | ret = -1; | |
1049 | goto error; | |
1050 | } | |
1051 | if (lttng_strncpy(msg.new_path, new_path, | |
1052 | sizeof(msg.new_path))) { | |
1053 | ret = -1; | |
1054 | goto error; | |
1055 | } | |
1056 | ||
1057 | /* Send command */ | |
1058 | ret = send_command(rsock, RELAYD_ROTATE_RENAME, (void *) &msg, sizeof(msg), 0); | |
1059 | if (ret < 0) { | |
1060 | goto error; | |
1061 | } | |
1062 | ||
1063 | /* Receive response */ | |
1064 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); | |
1065 | if (ret < 0) { | |
1066 | goto error; | |
1067 | } | |
1068 | ||
1069 | reply.ret_code = be32toh(reply.ret_code); | |
1070 | ||
1071 | /* Return session id or negative ret code. */ | |
1072 | if (reply.ret_code != LTTNG_OK) { | |
1073 | ret = -1; | |
1074 | ERR("Relayd rotate rename replied error %d", reply.ret_code); | |
1075 | } else { | |
1076 | /* Success */ | |
1077 | ret = 0; | |
1078 | } | |
1079 | ||
1080 | DBG("Relayd rotate rename completed successfully"); | |
1081 | ||
d26a0e48 JD |
1082 | error: |
1083 | return ret; | |
1084 | ||
1085 | } |