?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
lib: Split out write_data[_iov]
[Samba.git] / source3 / lib / ctdbd_conn.c
blobc41ec530a7b336727e17d8f76e37430ea16dc81b
1 /*
2 Unix SMB/CIFS implementation.
3 Samba internal messaging functions
4 Copyright (C) 2007 by Volker Lendecke
5 Copyright (C) 2007 by Andrew Tridgell
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 as published by
9 the Free Software Foundation; either version 3 of the License, or
10 (at your option) any later version.
11
12 This program is distributed in the hope that it will be useful,
13 but WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 GNU General Public License for more details.
16
17 You should have received a copy of the GNU General Public License
18 along with this program. If not, see <http://www.gnu.org/licenses/>.
19 */
20
21 #include "includes.h"
22 #include "util_tdb.h"
23 #include "serverid.h"
24 #include "ctdbd_conn.h"
25 #include "system/select.h"
26 #include "lib/sys_rw_data.h"
27
28 #include "messages.h"
29
30 /*
31 * It is not possible to include ctdb.h and tdb_compat.h (included via
32 * some other include above) without warnings. This fixes those
33 * warnings.
34 */
35
36 #ifdef typesafe_cb
37 #undef typesafe_cb
38 #endif
39
40 #ifdef typesafe_cb_preargs
41 #undef typesafe_cb_preargs
42 #endif
43
44 #ifdef typesafe_cb_postargs
45 #undef typesafe_cb_postargs
46 #endif
47
48 /* paths to these include files come from --with-ctdb= in configure */
49
50 #include "ctdb.h"
51 #include "ctdb_private.h"
52
53 struct ctdbd_connection {
54 struct messaging_context *msg_ctx;
55 uint32_t reqid;
56 uint32_t our_vnn;
57 uint64_t rand_srvid;
58 uint64_t *srvids;
59 int fd;
60 struct tevent_fd *fde;
61
62 bool (*release_ip_handler)(const char *ip_addr, void *private_data);
63 void *release_ip_priv;
64 };
65
66 static uint32_t ctdbd_next_reqid(struct ctdbd_connection *conn)
67 {
68 conn->reqid += 1;
69 if (conn->reqid == 0) {
70 conn->reqid += 1;
71 }
72 return conn->reqid;
73 }
74
75 static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
76 uint32_t vnn, uint32_t opcode,
77 uint64_t srvid, uint32_t flags, TDB_DATA data,
78 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
79 int *cstatus);
80
81 /*
82 * exit on fatal communications errors with the ctdbd daemon
83 */
84 static void cluster_fatal(const char *why)
85 {
86 DEBUG(0,("cluster fatal event: %s - exiting immediately\n", why));
87 /* we don't use smb_panic() as we don't want to delay to write
88 a core file. We need to release this process id immediately
89 so that someone else can take over without getting sharing
90 violations */
91 _exit(1);
92 }
93
94 /*
95 *
96 */
97 static void ctdb_packet_dump(struct ctdb_req_header *hdr)
98 {
99 if (DEBUGLEVEL < 11) {
100 return;
101 }
102 DEBUGADD(11, ("len=%d, magic=%x, vers=%d, gen=%d, op=%d, reqid=%d\n",
103 (int)hdr->length, (int)hdr->ctdb_magic,
104 (int)hdr->ctdb_version, (int)hdr->generation,
105 (int)hdr->operation, (int)hdr->reqid));
106 }
107
108 /*
109 * Register a srvid with ctdbd
110 */
111 NTSTATUS register_with_ctdbd(struct ctdbd_connection *conn, uint64_t srvid)
112 {
113
114 NTSTATUS status;
115 int cstatus;
116 size_t num_srvids;
117 uint64_t *tmp;
118
119 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
120 CTDB_CONTROL_REGISTER_SRVID, srvid, 0,
121 tdb_null, NULL, NULL, &cstatus);
122 if (!NT_STATUS_IS_OK(status)) {
123 return status;
124 }
125
126 num_srvids = talloc_array_length(conn->srvids);
127
128 tmp = talloc_realloc(conn, conn->srvids, uint64_t,
129 num_srvids + 1);
130 if (tmp == NULL) {
131 return NT_STATUS_NO_MEMORY;
132 }
133 conn->srvids = tmp;
134
135 conn->srvids[num_srvids] = srvid;
136 return NT_STATUS_OK;
137 }
138
139 static bool ctdb_is_our_srvid(struct ctdbd_connection *conn, uint64_t srvid)
140 {
141 size_t i, num_srvids;
142
143 num_srvids = talloc_array_length(conn->srvids);
144
145 for (i=0; i<num_srvids; i++) {
146 if (srvid == conn->srvids[i]) {
147 return true;
148 }
149 }
150 return false;
151 }
152
153 /*
154 * get our vnn from the cluster
155 */
156 static NTSTATUS get_cluster_vnn(struct ctdbd_connection *conn, uint32_t *vnn)
157 {
158 int32_t cstatus=-1;
159 NTSTATUS status;
160 status = ctdbd_control(conn,
161 CTDB_CURRENT_NODE, CTDB_CONTROL_GET_PNN, 0, 0,
162 tdb_null, NULL, NULL, &cstatus);
163 if (!NT_STATUS_IS_OK(status)) {
164 DEBUG(1, ("ctdbd_control failed: %s\n", nt_errstr(status)));
165 return status;
166 }
167 *vnn = (uint32_t)cstatus;
168 return status;
169 }
170
171 /*
172 * Are we active (i.e. not banned or stopped?)
173 */
174 static bool ctdbd_working(struct ctdbd_connection *conn, uint32_t vnn)
175 {
176 int32_t cstatus=-1;
177 NTSTATUS status;
178 TDB_DATA outdata;
179 struct ctdb_node_map *m;
180 uint32_t failure_flags;
181 bool ret = false;
182 int i;
183
184 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
185 CTDB_CONTROL_GET_NODEMAP, 0, 0,
186 tdb_null, talloc_tos(), &outdata, &cstatus);
187 if (!NT_STATUS_IS_OK(status)) {
188 DEBUG(1, ("ctdbd_control failed: %s\n", nt_errstr(status)));
189 return false;
190 }
191 if ((cstatus != 0) || (outdata.dptr == NULL)) {
192 DEBUG(2, ("Received invalid ctdb data\n"));
193 return false;
194 }
195
196 m = (struct ctdb_node_map *)outdata.dptr;
197
198 for (i=0; i<m->num; i++) {
199 if (vnn == m->nodes[i].pnn) {
200 break;
201 }
202 }
203
204 if (i == m->num) {
205 DEBUG(2, ("Did not find ourselves (node %d) in nodemap\n",
206 (int)vnn));
207 goto fail;
208 }
209
210 failure_flags = NODE_FLAGS_BANNED | NODE_FLAGS_DISCONNECTED
211 | NODE_FLAGS_PERMANENTLY_DISABLED | NODE_FLAGS_STOPPED;
212
213 if ((m->nodes[i].flags & failure_flags) != 0) {
214 DEBUG(2, ("Node has status %x, not active\n",
215 (int)m->nodes[i].flags));
216 goto fail;
217 }
218
219 ret = true;
220 fail:
221 TALLOC_FREE(outdata.dptr);
222 return ret;
223 }
224
225 uint32_t ctdbd_vnn(const struct ctdbd_connection *conn)
226 {
227 return conn->our_vnn;
228 }
229
230 const char *lp_ctdbd_socket(void)
231 {
232 const char *ret;
233
234 ret = lp__ctdbd_socket();
235 if (ret != NULL && strlen(ret) > 0) {
236 return ret;
237 }
238
239 return CTDB_SOCKET;
240 }
241
242 /*
243 * Get us a ctdb connection
244 */
245
246 static int ctdbd_connect(int *pfd)
247 {
248 const char *sockname = lp_ctdbd_socket();
249 struct sockaddr_un addr = { 0, };
250 int fd;
251 socklen_t salen;
252 size_t namelen;
253
254 fd = socket(AF_UNIX, SOCK_STREAM, 0);
255 if (fd == -1) {
256 int err = errno;
257 DEBUG(3, ("Could not create socket: %s\n", strerror(err)));
258 return err;
259 }
260
261 addr.sun_family = AF_UNIX;
262
263 namelen = strlcpy(addr.sun_path, sockname, sizeof(addr.sun_path));
264 if (namelen >= sizeof(addr.sun_path)) {
265 DEBUG(3, ("%s: Socket name too long: %s\n", __func__,
266 sockname));
267 close(fd);
268 return ENAMETOOLONG;
269 }
270
271 salen = sizeof(struct sockaddr_un);
272
273 if (connect(fd, (struct sockaddr *)(void *)&addr, salen) == -1) {
274 int err = errno;
275 DEBUG(1, ("connect(%s) failed: %s\n", sockname,
276 strerror(err)));
277 close(fd);
278 return err;
279 }
280
281 *pfd = fd;
282 return 0;
283 }
284
285 /*
286 * State necessary to defer an incoming message while we are waiting for a
287 * ctdb reply.
288 */
289
290 struct deferred_msg_state {
291 struct messaging_context *msg_ctx;
292 struct messaging_rec *rec;
293 };
294
295 /*
296 * Timed event handler for the deferred message
297 */
298
299 static void deferred_message_dispatch(struct tevent_context *event_ctx,
300 struct tevent_timer *te,
301 struct timeval now,
302 void *private_data)
303 {
304 struct deferred_msg_state *state = talloc_get_type_abort(
305 private_data, struct deferred_msg_state);
306
307 messaging_dispatch_rec(state->msg_ctx, state->rec);
308 TALLOC_FREE(state);
309 TALLOC_FREE(te);
310 }
311
312 /*
313 * Fetch a messaging_rec from an incoming ctdb style message
314 */
315
316 static struct messaging_rec *ctdb_pull_messaging_rec(TALLOC_CTX *mem_ctx,
317 size_t overall_length,
318 struct ctdb_req_message *msg)
319 {
320 struct messaging_rec *result;
321 DATA_BLOB blob;
322 enum ndr_err_code ndr_err;
323
324 if ((overall_length < offsetof(struct ctdb_req_message, data))
325 || (overall_length
326 < offsetof(struct ctdb_req_message, data) + msg->datalen)) {
327
328 cluster_fatal("got invalid msg length");
329 }
330
331 if (!(result = talloc(mem_ctx, struct messaging_rec))) {
332 DEBUG(0, ("talloc failed\n"));
333 return NULL;
334 }
335
336 blob = data_blob_const(msg->data, msg->datalen);
337
338 ndr_err = ndr_pull_struct_blob(
339 &blob, result, result,
340 (ndr_pull_flags_fn_t)ndr_pull_messaging_rec);
341
342 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
343 DEBUG(0, ("ndr_pull_struct_blob failed: %s\n",
344 ndr_errstr(ndr_err)));
345 TALLOC_FREE(result);
346 return NULL;
347 }
348
349 if (DEBUGLEVEL >= 11) {
350 DEBUG(11, ("ctdb_pull_messaging_rec:\n"));
351 NDR_PRINT_DEBUG(messaging_rec, result);
352 }
353
354 return result;
355 }
356
357 static NTSTATUS ctdb_read_packet(int fd, TALLOC_CTX *mem_ctx,
358 struct ctdb_req_header **result)
359 {
360 int timeout = lp_ctdb_timeout();
361 struct ctdb_req_header *req;
362 int ret, revents;
363 uint32_t msglen;
364 NTSTATUS status;
365
366 if (timeout == 0) {
367 timeout = -1;
368 }
369
370 if (timeout != -1) {
371 ret = poll_one_fd(fd, POLLIN, timeout, &revents);
372 if (ret == -1) {
373 return map_nt_error_from_unix(errno);
374 }
375 if (ret == 0) {
376 return NT_STATUS_IO_TIMEOUT;
377 }
378 if (ret != 1) {
379 return NT_STATUS_UNEXPECTED_IO_ERROR;
380 }
381 }
382
383 status = read_data_ntstatus(fd, (char *)&msglen, sizeof(msglen));
384 if (!NT_STATUS_IS_OK(status)) {
385 return status;
386 }
387
388 if (msglen < sizeof(struct ctdb_req_header)) {
389 return NT_STATUS_UNEXPECTED_IO_ERROR;
390 }
391
392 req = talloc_size(mem_ctx, msglen);
393 if (req == NULL) {
394 return NT_STATUS_NO_MEMORY;
395 }
396 talloc_set_name_const(req, "struct ctdb_req_header");
397
398 req->length = msglen;
399
400 status = read_data_ntstatus(fd, ((char *)req) + sizeof(msglen),
401 msglen - sizeof(msglen));
402 if (!NT_STATUS_IS_OK(status)) {
403 return status;
404 }
405
406 *result = req;
407 return NT_STATUS_OK;
408 }
409
410 /*
411 * Read a full ctdbd request. If we have a messaging context, defer incoming
412 * messages that might come in between.
413 */
414
415 static NTSTATUS ctdb_read_req(struct ctdbd_connection *conn, uint32_t reqid,
416 TALLOC_CTX *mem_ctx,
417 struct ctdb_req_header **result)
418 {
419 struct ctdb_req_header *hdr;
420 NTSTATUS status;
421
422 next_pkt:
423
424 status = ctdb_read_packet(conn->fd, mem_ctx, &hdr);
425 if (!NT_STATUS_IS_OK(status)) {
426 DEBUG(0, ("ctdb_read_packet failed: %s\n", nt_errstr(status)));
427 cluster_fatal("ctdbd died\n");
428 }
429
430 DEBUG(11, ("Received ctdb packet\n"));
431 ctdb_packet_dump(hdr);
432
433 if (hdr->operation == CTDB_REQ_MESSAGE) {
434 struct tevent_timer *evt;
435 struct deferred_msg_state *msg_state;
436 struct ctdb_req_message *msg = (struct ctdb_req_message *)hdr;
437
438 if (conn->msg_ctx == NULL) {
439 DEBUG(1, ("Got a message without having a msg ctx, "
440 "dropping msg %llu\n",
441 (long long unsigned)msg->srvid));
442 goto next_pkt;
443 }
444
445 if ((conn->release_ip_handler != NULL)
446 && (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
447 bool ret;
448
449 /* must be dispatched immediately */
450 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
451 ret = conn->release_ip_handler((const char *)msg->data,
452 conn->release_ip_priv);
453 TALLOC_FREE(hdr);
454
455 if (ret) {
456 /*
457 * We need to release the ip,
458 * so return an error to the upper layers.
459 *
460 * We make sure we don't trigger this again.
461 */
462 conn->release_ip_handler = NULL;
463 conn->release_ip_priv = NULL;
464 return NT_STATUS_ADDRESS_CLOSED;
465 }
466 goto next_pkt;
467 }
468
469 if ((msg->srvid == CTDB_SRVID_RECONFIGURE)
470 || (msg->srvid == CTDB_SRVID_SAMBA_NOTIFY)) {
471
472 DEBUG(1, ("ctdb_read_req: Got %s message\n",
473 (msg->srvid == CTDB_SRVID_RECONFIGURE)
474 ? "cluster reconfigure" : "SAMBA_NOTIFY"));
475
476 messaging_send(conn->msg_ctx,
477 messaging_server_id(conn->msg_ctx),
478 MSG_SMB_BRL_VALIDATE, &data_blob_null);
479 TALLOC_FREE(hdr);
480 goto next_pkt;
481 }
482
483 msg_state = talloc(NULL, struct deferred_msg_state);
484 if (msg_state == NULL) {
485 DEBUG(0, ("talloc failed\n"));
486 TALLOC_FREE(hdr);
487 goto next_pkt;
488 }
489
490 if (!(msg_state->rec = ctdb_pull_messaging_rec(
491 msg_state, msg->hdr.length, msg))) {
492 DEBUG(0, ("ctdbd_pull_messaging_rec failed\n"));
493 TALLOC_FREE(msg_state);
494 TALLOC_FREE(hdr);
495 goto next_pkt;
496 }
497
498 TALLOC_FREE(hdr);
499
500 msg_state->msg_ctx = conn->msg_ctx;
501
502 /*
503 * We're waiting for a call reply, but an async message has
504 * crossed. Defer dispatching to the toplevel event loop.
505 */
506 evt = tevent_add_timer(messaging_tevent_context(conn->msg_ctx),
507 messaging_tevent_context(conn->msg_ctx),
508 timeval_zero(),
509 deferred_message_dispatch,
510 msg_state);
511 if (evt == NULL) {
512 DEBUG(0, ("event_add_timed failed\n"));
513 TALLOC_FREE(msg_state);
514 TALLOC_FREE(hdr);
515 goto next_pkt;
516 }
517
518 goto next_pkt;
519 }
520
521 if ((reqid != 0) && (hdr->reqid != reqid)) {
522 /* we got the wrong reply */
523 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
524 "been %u\n", hdr->reqid, reqid));
525 TALLOC_FREE(hdr);
526 goto next_pkt;
527 }
528
529 *result = talloc_move(mem_ctx, &hdr);
530
531 return NT_STATUS_OK;
532 }
533
534 static int ctdbd_connection_destructor(struct ctdbd_connection *c)
535 {
536 close(c->fd);
537 return 0;
538 }
539 /*
540 * Get us a ctdbd connection
541 */
542
543 static NTSTATUS ctdbd_init_connection(TALLOC_CTX *mem_ctx,
544 struct ctdbd_connection **pconn)
545 {
546 struct ctdbd_connection *conn;
547 int ret;
548 NTSTATUS status;
549
550 if (!(conn = talloc_zero(mem_ctx, struct ctdbd_connection))) {
551 DEBUG(0, ("talloc failed\n"));
552 return NT_STATUS_NO_MEMORY;
553 }
554
555 ret = ctdbd_connect(&conn->fd);
556 if (ret != 0) {
557 status = map_nt_error_from_unix(errno);
558 DEBUG(10, ("ctdbd_connect failed: %s\n", strerror(errno)));
559 goto fail;
560 }
561 talloc_set_destructor(conn, ctdbd_connection_destructor);
562
563 status = get_cluster_vnn(conn, &conn->our_vnn);
564
565 if (!NT_STATUS_IS_OK(status)) {
566 DEBUG(10, ("get_cluster_vnn failed: %s\n", nt_errstr(status)));
567 goto fail;
568 }
569
570 if (!ctdbd_working(conn, conn->our_vnn)) {
571 DEBUG(2, ("Node is not working, can not connect\n"));
572 status = NT_STATUS_INTERNAL_DB_ERROR;
573 goto fail;
574 }
575
576 generate_random_buffer((unsigned char *)&conn->rand_srvid,
577 sizeof(conn->rand_srvid));
578
579 status = register_with_ctdbd(conn, conn->rand_srvid);
580
581 if (!NT_STATUS_IS_OK(status)) {
582 DEBUG(5, ("Could not register random srvid: %s\n",
583 nt_errstr(status)));
584 goto fail;
585 }
586
587 *pconn = conn;
588 return NT_STATUS_OK;
589
590 fail:
591 TALLOC_FREE(conn);
592 return status;
593 }
594
595 /*
596 * Get us a ctdbd connection and register us as a process
597 */
598
599 NTSTATUS ctdbd_messaging_connection(TALLOC_CTX *mem_ctx,
600 struct ctdbd_connection **pconn)
601 {
602 struct ctdbd_connection *conn;
603 NTSTATUS status;
604
605 status = ctdbd_init_connection(mem_ctx, &conn);
606
607 if (!NT_STATUS_IS_OK(status)) {
608 return status;
609 }
610
611 status = register_with_ctdbd(conn, (uint64_t)getpid());
612 if (!NT_STATUS_IS_OK(status)) {
613 goto fail;
614 }
615
616 status = register_with_ctdbd(conn, MSG_SRVID_SAMBA);
617 if (!NT_STATUS_IS_OK(status)) {
618 goto fail;
619 }
620
621 status = register_with_ctdbd(conn, CTDB_SRVID_SAMBA_NOTIFY);
622 if (!NT_STATUS_IS_OK(status)) {
623 goto fail;
624 }
625
626 *pconn = conn;
627 return NT_STATUS_OK;
628
629 fail:
630 TALLOC_FREE(conn);
631 return status;
632 }
633
634 struct messaging_context *ctdb_conn_msg_ctx(struct ctdbd_connection *conn)
635 {
636 return conn->msg_ctx;
637 }
638
639 int ctdbd_conn_get_fd(struct ctdbd_connection *conn)
640 {
641 return conn->fd;
642 }
643
644 /*
645 * Packet handler to receive and handle a ctdb message
646 */
647 static NTSTATUS ctdb_handle_message(struct messaging_context *msg_ctx,
648 struct ctdbd_connection *conn,
649 struct ctdb_req_header *hdr)
650 {
651 struct ctdb_req_message *msg;
652 struct messaging_rec *msg_rec;
653
654 if (hdr->operation != CTDB_REQ_MESSAGE) {
655 DEBUG(0, ("Received async msg of type %u, discarding\n",
656 hdr->operation));
657 return NT_STATUS_INVALID_PARAMETER;
658 }
659
660 msg = (struct ctdb_req_message *)hdr;
661
662 if ((conn->release_ip_handler != NULL)
663 && (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
664 bool ret;
665
666 /* must be dispatched immediately */
667 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
668 ret = conn->release_ip_handler((const char *)msg->data,
669 conn->release_ip_priv);
670 if (ret) {
671 /*
672 * We need to release the ip.
673 *
674 * We make sure we don't trigger this again.
675 */
676 conn->release_ip_handler = NULL;
677 conn->release_ip_priv = NULL;
678 }
679 return NT_STATUS_OK;
680 }
681
682 SMB_ASSERT(conn->msg_ctx != NULL);
683
684 if ((msg->srvid == CTDB_SRVID_RECONFIGURE)
685 || (msg->srvid == CTDB_SRVID_SAMBA_NOTIFY)){
686 DEBUG(0,("Got cluster reconfigure message\n"));
687 /*
688 * when the cluster is reconfigured or someone of the
689 * family has passed away (SAMBA_NOTIFY), we need to
690 * clean the brl database
691 */
692 messaging_send(conn->msg_ctx,
693 messaging_server_id(conn->msg_ctx),
694 MSG_SMB_BRL_VALIDATE, &data_blob_null);
695
696 return NT_STATUS_OK;
697 }
698
699 if (!ctdb_is_our_srvid(conn, msg->srvid)) {
700 DEBUG(0,("Got unexpected message with srvid=%llu\n",
701 (unsigned long long)msg->srvid));
702 return NT_STATUS_OK;
703 }
704
705 msg_rec = ctdb_pull_messaging_rec(talloc_tos(), msg->hdr.length, msg);
706 if (msg_rec == NULL) {
707 DEBUG(10, ("ctdb_pull_messaging_rec failed\n"));
708 return NT_STATUS_NO_MEMORY;
709 }
710 messaging_dispatch_rec(conn->msg_ctx, msg_rec);
711 return NT_STATUS_OK;
712 }
713
714 /*
715 * The ctdbd socket is readable asynchronuously
716 */
717
718 static void ctdbd_socket_handler(struct tevent_context *event_ctx,
719 struct tevent_fd *event,
720 uint16 flags,
721 void *private_data)
722 {
723 struct ctdbd_connection *conn = talloc_get_type_abort(
724 private_data, struct ctdbd_connection);
725 struct ctdb_req_header *hdr;
726 NTSTATUS status;
727
728 status = ctdb_read_packet(conn->fd, talloc_tos(), &hdr);
729 if (!NT_STATUS_IS_OK(status)) {
730 DEBUG(0, ("ctdb_read_packet failed: %s\n", nt_errstr(status)));
731 cluster_fatal("ctdbd died\n");
732 }
733
734 status = ctdb_handle_message(conn->msg_ctx, conn, hdr);
735 if (!NT_STATUS_IS_OK(status)) {
736 DEBUG(10, ("could not handle incoming message: %s\n",
737 nt_errstr(status)));
738 }
739 }
740
741 /*
742 * Prepare a ctdbd connection to receive messages
743 */
744
745 NTSTATUS ctdbd_register_msg_ctx(struct ctdbd_connection *conn,
746 struct messaging_context *msg_ctx)
747 {
748 SMB_ASSERT(conn->msg_ctx == NULL);
749 SMB_ASSERT(conn->fde == NULL);
750
751 if (!(conn->fde = tevent_add_fd(messaging_tevent_context(msg_ctx),
752 conn,
753 conn->fd,
754 TEVENT_FD_READ,
755 ctdbd_socket_handler,
756 conn))) {
757 DEBUG(0, ("event_add_fd failed\n"));
758 return NT_STATUS_NO_MEMORY;
759 }
760
761 conn->msg_ctx = msg_ctx;
762
763 return NT_STATUS_OK;
764 }
765
766 /*
767 * Send a messaging message across a ctdbd
768 */
769
770 NTSTATUS ctdbd_messaging_send(struct ctdbd_connection *conn,
771 uint32_t dst_vnn, uint64_t dst_srvid,
772 struct messaging_rec *msg)
773 {
774 DATA_BLOB blob;
775 NTSTATUS status;
776 enum ndr_err_code ndr_err;
777
778 ndr_err = ndr_push_struct_blob(
779 &blob, talloc_tos(), msg,
780 (ndr_push_flags_fn_t)ndr_push_messaging_rec);
781
782 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
783 DEBUG(0, ("ndr_push_struct_blob failed: %s\n",
784 ndr_errstr(ndr_err)));
785 return ndr_map_error2ntstatus(ndr_err);
786 }
787
788 status = ctdbd_messaging_send_blob(conn, dst_vnn, dst_srvid,
789 blob.data, blob.length);
790 TALLOC_FREE(blob.data);
791 return status;
792 }
793
794 NTSTATUS ctdbd_messaging_send_blob(struct ctdbd_connection *conn,
795 uint32_t dst_vnn, uint64_t dst_srvid,
796 const uint8_t *buf, size_t buflen)
797 {
798 struct ctdb_req_message r;
799 struct iovec iov[2];
800 ssize_t nwritten;
801
802 r.hdr.length = offsetof(struct ctdb_req_message, data) + buflen;
803 r.hdr.ctdb_magic = CTDB_MAGIC;
804 r.hdr.ctdb_version = CTDB_PROTOCOL;
805 r.hdr.generation = 1;
806 r.hdr.operation = CTDB_REQ_MESSAGE;
807 r.hdr.destnode = dst_vnn;
808 r.hdr.srcnode = conn->our_vnn;
809 r.hdr.reqid = 0;
810 r.srvid = dst_srvid;
811 r.datalen = buflen;
812
813 DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
814 ctdb_packet_dump(&r.hdr);
815
816 iov[0].iov_base = &r;
817 iov[0].iov_len = offsetof(struct ctdb_req_message, data);
818 iov[1].iov_base = discard_const_p(uint8_t, buf);
819 iov[1].iov_len = buflen;
820
821 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
822 if (nwritten == -1) {
823 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
824 cluster_fatal("cluster dispatch daemon msg write error\n");
825 }
826
827 return NT_STATUS_OK;
828 }
829
830 /*
831 * send/recv a generic ctdb control message
832 */
833 static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
834 uint32_t vnn, uint32_t opcode,
835 uint64_t srvid, uint32_t flags,
836 TDB_DATA data,
837 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
838 int *cstatus)
839 {
840 struct ctdb_req_control req;
841 struct ctdb_req_header *hdr;
842 struct ctdb_reply_control *reply = NULL;
843 struct ctdbd_connection *new_conn = NULL;
844 struct iovec iov[2];
845 ssize_t nwritten;
846 NTSTATUS status;
847
848 if (conn == NULL) {
849 status = ctdbd_init_connection(NULL, &new_conn);
850
851 if (!NT_STATUS_IS_OK(status)) {
852 DEBUG(10, ("Could not init temp connection: %s\n",
853 nt_errstr(status)));
854 goto fail;
855 }
856
857 conn = new_conn;
858 }
859
860 ZERO_STRUCT(req);
861 req.hdr.length = offsetof(struct ctdb_req_control, data) + data.dsize;
862 req.hdr.ctdb_magic = CTDB_MAGIC;
863 req.hdr.ctdb_version = CTDB_PROTOCOL;
864 req.hdr.operation = CTDB_REQ_CONTROL;
865 req.hdr.reqid = ctdbd_next_reqid(conn);
866 req.hdr.destnode = vnn;
867 req.opcode = opcode;
868 req.srvid = srvid;
869 req.datalen = data.dsize;
870 req.flags = flags;
871
872 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
873 ctdb_packet_dump(&req.hdr);
874
875 iov[0].iov_base = &req;
876 iov[0].iov_len = offsetof(struct ctdb_req_control, data);
877 iov[1].iov_base = data.dptr;
878 iov[1].iov_len = data.dsize;
879
880 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
881 if (nwritten == -1) {
882 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
883 cluster_fatal("cluster dispatch daemon msg write error\n");
884 }
885
886 if (flags & CTDB_CTRL_FLAG_NOREPLY) {
887 TALLOC_FREE(new_conn);
888 if (cstatus) {
889 *cstatus = 0;
890 }
891 return NT_STATUS_OK;
892 }
893
894 status = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
895
896 if (!NT_STATUS_IS_OK(status)) {
897 DEBUG(10, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
898 goto fail;
899 }
900
901 if (hdr->operation != CTDB_REPLY_CONTROL) {
902 DEBUG(0, ("received invalid reply\n"));
903 goto fail;
904 }
905 reply = (struct ctdb_reply_control *)hdr;
906
907 if (outdata) {
908 if (!(outdata->dptr = (uint8 *)talloc_memdup(
909 mem_ctx, reply->data, reply->datalen))) {
910 TALLOC_FREE(reply);
911 return NT_STATUS_NO_MEMORY;
912 }
913 outdata->dsize = reply->datalen;
914 }
915 if (cstatus) {
916 (*cstatus) = reply->status;
917 }
918
919 status = NT_STATUS_OK;
920
921 fail:
922 TALLOC_FREE(new_conn);
923 TALLOC_FREE(reply);
924 return status;
925 }
926
927 /*
928 * see if a remote process exists
929 */
930 bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32_t vnn, pid_t pid)
931 {
932 struct server_id id;
933 bool result;
934
935 id.pid = pid;
936 id.vnn = vnn;
937
938 if (!ctdb_processes_exist(conn, &id, 1, &result)) {
939 DEBUG(10, ("ctdb_processes_exist failed\n"));
940 return false;
941 }
942 return result;
943 }
944
945 bool ctdb_processes_exist(struct ctdbd_connection *conn,
946 const struct server_id *pids, int num_pids,
947 bool *results)
948 {
949 TALLOC_CTX *frame = talloc_stackframe();
950 int i, num_received;
951 NTSTATUS status;
952 uint32_t *reqids;
953 bool result = false;
954
955 reqids = talloc_array(talloc_tos(), uint32_t, num_pids);
956 if (reqids == NULL) {
957 goto fail;
958 }
959
960 for (i=0; i<num_pids; i++) {
961 struct ctdb_req_control req;
962 pid_t pid;
963 struct iovec iov[2];
964 ssize_t nwritten;
965
966 results[i] = false;
967 reqids[i] = ctdbd_next_reqid(conn);
968
969 ZERO_STRUCT(req);
970
971 /*
972 * pids[i].pid is uint64_t, scale down to pid_t which
973 * is the wire protocol towards ctdb.
974 */
975 pid = pids[i].pid;
976
977 DEBUG(10, ("Requesting PID %d/%d, reqid=%d\n",
978 (int)pids[i].vnn, (int)pid,
979 (int)reqids[i]));
980
981 req.hdr.length = offsetof(struct ctdb_req_control, data);
982 req.hdr.length += sizeof(pid);
983 req.hdr.ctdb_magic = CTDB_MAGIC;
984 req.hdr.ctdb_version = CTDB_PROTOCOL;
985 req.hdr.operation = CTDB_REQ_CONTROL;
986 req.hdr.reqid = reqids[i];
987 req.hdr.destnode = pids[i].vnn;
988 req.opcode = CTDB_CONTROL_PROCESS_EXISTS;
989 req.srvid = 0;
990 req.datalen = sizeof(pid);
991 req.flags = 0;
992
993 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
994 ctdb_packet_dump(&req.hdr);
995
996 iov[0].iov_base = &req;
997 iov[0].iov_len = offsetof(struct ctdb_req_control, data);
998 iov[1].iov_base = &pid;
999 iov[1].iov_len = sizeof(pid);
1000
1001 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
1002 if (nwritten == -1) {
1003 status = map_nt_error_from_unix(errno);
1004 DEBUG(10, ("write_data_iov failed: %s\n",
1005 strerror(errno)));
1006 goto fail;
1007 }
1008 }
1009
1010 num_received = 0;
1011
1012 while (num_received < num_pids) {
1013 struct ctdb_req_header *hdr;
1014 struct ctdb_reply_control *reply;
1015 uint32_t reqid;
1016
1017 status = ctdb_read_req(conn, 0, talloc_tos(), &hdr);
1018 if (!NT_STATUS_IS_OK(status)) {
1019 DEBUG(10, ("ctdb_read_req failed: %s\n",
1020 nt_errstr(status)));
1021 goto fail;
1022 }
1023
1024 if (hdr->operation != CTDB_REPLY_CONTROL) {
1025 DEBUG(10, ("Received invalid reply\n"));
1026 goto fail;
1027 }
1028 reply = (struct ctdb_reply_control *)hdr;
1029
1030 reqid = reply->hdr.reqid;
1031
1032 DEBUG(10, ("Received reqid %d\n", (int)reqid));
1033
1034 for (i=0; i<num_pids; i++) {
1035 if (reqid == reqids[i]) {
1036 break;
1037 }
1038 }
1039 if (i == num_pids) {
1040 DEBUG(10, ("Received unknown record number %u\n",
1041 (unsigned)reqid));
1042 goto fail;
1043 }
1044 results[i] = ((reply->status) == 0);
1045 TALLOC_FREE(reply);
1046 num_received += 1;
1047 }
1048
1049 result = true;
1050 fail:
1051 TALLOC_FREE(frame);
1052 return result;
1053 }
1054
1055 struct ctdb_vnn_list {
1056 uint32_t vnn;
1057 uint32_t reqid;
1058 unsigned num_srvids;
1059 unsigned num_filled;
1060 uint64_t *srvids;
1061 unsigned *pid_indexes;
1062 };
1063
1064 /*
1065 * Get a list of all vnns mentioned in a list of
1066 * server_ids. vnn_indexes tells where in the vnns array we have to
1067 * place the pids.
1068 */
1069 static bool ctdb_collect_vnns(TALLOC_CTX *mem_ctx,
1070 const struct server_id *pids, unsigned num_pids,
1071 struct ctdb_vnn_list **pvnns,
1072 unsigned *pnum_vnns)
1073 {
1074 struct ctdb_vnn_list *vnns = NULL;
1075 unsigned *vnn_indexes = NULL;
1076 unsigned i, num_vnns = 0;
1077
1078 vnn_indexes = talloc_array(mem_ctx, unsigned, num_pids);
1079 if (vnn_indexes == NULL) {
1080 DEBUG(1, ("talloc_array failed\n"));
1081 goto fail;
1082 }
1083
1084 for (i=0; i<num_pids; i++) {
1085 unsigned j;
1086 uint32_t vnn = pids[i].vnn;
1087
1088 for (j=0; j<num_vnns; j++) {
1089 if (vnn == vnns[j].vnn) {
1090 break;
1091 }
1092 }
1093 vnn_indexes[i] = j;
1094
1095 if (j < num_vnns) {
1096 /*
1097 * Already in the array
1098 */
1099 vnns[j].num_srvids += 1;
1100 continue;
1101 }
1102 vnns = talloc_realloc(mem_ctx, vnns, struct ctdb_vnn_list,
1103 num_vnns+1);
1104 if (vnns == NULL) {
1105 DEBUG(1, ("talloc_realloc failed\n"));
1106 goto fail;
1107 }
1108 vnns[num_vnns].vnn = vnn;
1109 vnns[num_vnns].num_srvids = 1;
1110 vnns[num_vnns].num_filled = 0;
1111 num_vnns += 1;
1112 }
1113 for (i=0; i<num_vnns; i++) {
1114 struct ctdb_vnn_list *vnn = &vnns[i];
1115
1116 vnn->srvids = talloc_array(vnns, uint64_t, vnn->num_srvids);
1117 if (vnn->srvids == NULL) {
1118 DEBUG(1, ("talloc_array failed\n"));
1119 goto fail;
1120 }
1121 vnn->pid_indexes = talloc_array(vnns, unsigned,
1122 vnn->num_srvids);
1123 if (vnn->pid_indexes == NULL) {
1124 DEBUG(1, ("talloc_array failed\n"));
1125 goto fail;
1126 }
1127 }
1128 for (i=0; i<num_pids; i++) {
1129 struct ctdb_vnn_list *vnn = &vnns[vnn_indexes[i]];
1130 vnn->srvids[vnn->num_filled] = pids[i].unique_id;
1131 vnn->pid_indexes[vnn->num_filled] = i;
1132 vnn->num_filled += 1;
1133 }
1134
1135 TALLOC_FREE(vnn_indexes);
1136 *pvnns = vnns;
1137 *pnum_vnns = num_vnns;
1138 return true;
1139 fail:
1140 TALLOC_FREE(vnns);
1141 TALLOC_FREE(vnn_indexes);
1142 return false;
1143 }
1144
1145 bool ctdb_serverids_exist_supported(struct ctdbd_connection *conn)
1146 {
1147 return true;
1148 }
1149
1150 bool ctdb_serverids_exist(struct ctdbd_connection *conn,
1151 const struct server_id *pids, unsigned num_pids,
1152 bool *results)
1153 {
1154 unsigned i, num_received;
1155 NTSTATUS status;
1156 struct ctdb_vnn_list *vnns = NULL;
1157 unsigned num_vnns;
1158
1159 if (!ctdb_collect_vnns(talloc_tos(), pids, num_pids,
1160 &vnns, &num_vnns)) {
1161 DEBUG(1, ("ctdb_collect_vnns failed\n"));
1162 goto fail;
1163 }
1164
1165 for (i=0; i<num_vnns; i++) {
1166 struct ctdb_vnn_list *vnn = &vnns[i];
1167 struct ctdb_req_control req;
1168 struct iovec iov[2];
1169 ssize_t nwritten;
1170
1171 vnn->reqid = ctdbd_next_reqid(conn);
1172
1173 ZERO_STRUCT(req);
1174
1175 DEBUG(10, ("Requesting VNN %d, reqid=%d, num_srvids=%u\n",
1176 (int)vnn->vnn, (int)vnn->reqid, vnn->num_srvids));
1177
1178 req.hdr.length = offsetof(struct ctdb_req_control, data);
1179 req.hdr.ctdb_magic = CTDB_MAGIC;
1180 req.hdr.ctdb_version = CTDB_PROTOCOL;
1181 req.hdr.operation = CTDB_REQ_CONTROL;
1182 req.hdr.reqid = vnn->reqid;
1183 req.hdr.destnode = vnn->vnn;
1184 req.opcode = CTDB_CONTROL_CHECK_SRVIDS;
1185 req.srvid = 0;
1186 req.datalen = sizeof(uint64_t) * vnn->num_srvids;
1187 req.hdr.length += req.datalen;
1188 req.flags = 0;
1189
1190 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
1191 ctdb_packet_dump(&req.hdr);
1192
1193 iov[0].iov_base = &req;
1194 iov[0].iov_len = offsetof(struct ctdb_req_control, data);
1195 iov[1].iov_base = vnn->srvids;
1196 iov[1].iov_len = req.datalen;
1197
1198 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
1199 if (nwritten == -1) {
1200 status = map_nt_error_from_unix(errno);
1201 DEBUG(10, ("write_data_iov failed: %s\n",
1202 strerror(errno)));
1203 goto fail;
1204 }
1205 }
1206
1207 num_received = 0;
1208
1209 while (num_received < num_vnns) {
1210 struct ctdb_req_header *hdr;
1211 struct ctdb_reply_control *reply;
1212 struct ctdb_vnn_list *vnn;
1213 uint32_t reqid;
1214 uint8_t *reply_data;
1215
1216 status = ctdb_read_req(conn, 0, talloc_tos(), &hdr);
1217 if (!NT_STATUS_IS_OK(status)) {
1218 DEBUG(1, ("ctdb_read_req failed: %s\n",
1219 nt_errstr(status)));
1220 goto fail;
1221 }
1222
1223 if (hdr->operation != CTDB_REPLY_CONTROL) {
1224 DEBUG(1, ("Received invalid reply %u\n",
1225 (unsigned)reply->hdr.operation));
1226 goto fail;
1227 }
1228 reply = (struct ctdb_reply_control *)hdr;
1229
1230 reqid = reply->hdr.reqid;
1231
1232 DEBUG(10, ("Received reqid %d\n", (int)reqid));
1233
1234 for (i=0; i<num_vnns; i++) {
1235 if (reqid == vnns[i].reqid) {
1236 break;
1237 }
1238 }
1239 if (i == num_vnns) {
1240 DEBUG(1, ("Received unknown reqid number %u\n",
1241 (unsigned)reqid));
1242 goto fail;
1243 }
1244
1245 DEBUG(10, ("Found index %u\n", i));
1246
1247 vnn = &vnns[i];
1248
1249 DEBUG(10, ("Received vnn %u, vnn->num_srvids %u, datalen %u\n",
1250 (unsigned)vnn->vnn, vnn->num_srvids,
1251 (unsigned)reply->datalen));
1252
1253 if (reply->datalen >= ((vnn->num_srvids+7)/8)) {
1254 /*
1255 * Got a real reply
1256 */
1257 reply_data = reply->data;
1258 } else {
1259 /*
1260 * Got an error reply
1261 */
1262 DEBUG(5, ("Received short reply len %d, status %u, "
1263 "errorlen %u\n",
1264 (unsigned)reply->datalen,
1265 (unsigned)reply->status,
1266 (unsigned)reply->errorlen));
1267 dump_data(5, reply->data, reply->errorlen);
1268
1269 /*
1270 * This will trigger everything set to false
1271 */
1272 reply_data = NULL;
1273 }
1274
1275 for (i=0; i<vnn->num_srvids; i++) {
1276 int idx = vnn->pid_indexes[i];
1277
1278 if (pids[i].unique_id ==
1279 SERVERID_UNIQUE_ID_NOT_TO_VERIFY) {
1280 results[idx] = true;
1281 continue;
1282 }
1283 results[idx] =
1284 (reply_data != NULL) &&
1285 ((reply_data[i/8] & (1<<(i%8))) != 0);
1286 }
1287
1288 TALLOC_FREE(reply);
1289 num_received += 1;
1290 }
1291
1292 TALLOC_FREE(vnns);
1293 return true;
1294 fail:
1295 cluster_fatal("serverids_exist failed");
1296 return false;
1297 }
1298
1299 /*
1300 * Get a db path
1301 */
1302 char *ctdbd_dbpath(struct ctdbd_connection *conn,
1303 TALLOC_CTX *mem_ctx, uint32_t db_id)
1304 {
1305 NTSTATUS status;
1306 TDB_DATA data;
1307 int32_t cstatus;
1308
1309 data.dptr = (uint8_t*)&db_id;
1310 data.dsize = sizeof(db_id);
1311
1312 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1313 CTDB_CONTROL_GETDBPATH, 0, 0, data,
1314 mem_ctx, &data, &cstatus);
1315 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
1316 DEBUG(0,(__location__ " ctdb_control for getdbpath failed\n"));
1317 return NULL;
1318 }
1319
1320 return (char *)data.dptr;
1321 }
1322
1323 /*
1324 * attach to a ctdb database
1325 */
1326 NTSTATUS ctdbd_db_attach(struct ctdbd_connection *conn,
1327 const char *name, uint32_t *db_id, int tdb_flags)
1328 {
1329 NTSTATUS status;
1330 TDB_DATA data;
1331 int32_t cstatus;
1332 bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
1333
1334 data = string_term_tdb_data(name);
1335
1336 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1337 persistent
1338 ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
1339 : CTDB_CONTROL_DB_ATTACH,
1340 tdb_flags, 0, data, NULL, &data, &cstatus);
1341 if (!NT_STATUS_IS_OK(status)) {
1342 DEBUG(0, (__location__ " ctdb_control for db_attach "
1343 "failed: %s\n", nt_errstr(status)));
1344 return status;
1345 }
1346
1347 if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
1348 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
1349 return NT_STATUS_INTERNAL_ERROR;
1350 }
1351
1352 *db_id = *(uint32_t *)data.dptr;
1353 talloc_free(data.dptr);
1354
1355 if (!(tdb_flags & TDB_SEQNUM)) {
1356 return NT_STATUS_OK;
1357 }
1358
1359 data.dptr = (uint8_t *)db_id;
1360 data.dsize = sizeof(*db_id);
1361
1362 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1363 CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
1364 NULL, NULL, &cstatus);
1365 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
1366 DEBUG(0,(__location__ " ctdb_control for enable seqnum "
1367 "failed\n"));
1368 return NT_STATUS_IS_OK(status) ? NT_STATUS_INTERNAL_ERROR :
1369 status;
1370 }
1371
1372 return NT_STATUS_OK;
1373 }
1374
1375 /*
1376 * force the migration of a record to this node
1377 */
1378 NTSTATUS ctdbd_migrate(struct ctdbd_connection *conn, uint32_t db_id,
1379 TDB_DATA key)
1380 {
1381 struct ctdb_req_call req;
1382 struct ctdb_req_header *hdr;
1383 struct iovec iov[2];
1384 ssize_t nwritten;
1385 NTSTATUS status;
1386
1387 ZERO_STRUCT(req);
1388
1389 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
1390 req.hdr.ctdb_magic = CTDB_MAGIC;
1391 req.hdr.ctdb_version = CTDB_PROTOCOL;
1392 req.hdr.operation = CTDB_REQ_CALL;
1393 req.hdr.reqid = ctdbd_next_reqid(conn);
1394 req.flags = CTDB_IMMEDIATE_MIGRATION;
1395 req.callid = CTDB_NULL_FUNC;
1396 req.db_id = db_id;
1397 req.keylen = key.dsize;
1398
1399 DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
1400 ctdb_packet_dump(&req.hdr);
1401
1402 iov[0].iov_base = &req;
1403 iov[0].iov_len = offsetof(struct ctdb_req_call, data);
1404 iov[1].iov_base = key.dptr;
1405 iov[1].iov_len = key.dsize;
1406
1407 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
1408 if (nwritten == -1) {
1409 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
1410 cluster_fatal("cluster dispatch daemon msg write error\n");
1411 }
1412
1413 status = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
1414
1415 if (!NT_STATUS_IS_OK(status)) {
1416 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
1417 goto fail;
1418 }
1419
1420 if (hdr->operation != CTDB_REPLY_CALL) {
1421 DEBUG(0, ("received invalid reply\n"));
1422 status = NT_STATUS_INTERNAL_ERROR;
1423 goto fail;
1424 }
1425
1426 status = NT_STATUS_OK;
1427 fail:
1428
1429 TALLOC_FREE(hdr);
1430 return status;
1431 }
1432
1433 /*
1434 * Fetch a record and parse it
1435 */
1436 NTSTATUS ctdbd_parse(struct ctdbd_connection *conn, uint32_t db_id,
1437 TDB_DATA key, bool local_copy,
1438 void (*parser)(TDB_DATA key, TDB_DATA data,
1439 void *private_data),
1440 void *private_data)
1441 {
1442 struct ctdb_req_call req;
1443 struct ctdb_req_header *hdr = NULL;
1444 struct ctdb_reply_call *reply;
1445 struct iovec iov[2];
1446 ssize_t nwritten;
1447 NTSTATUS status;
1448 uint32_t flags;
1449
1450 flags = local_copy ? CTDB_WANT_READONLY : 0;
1451
1452 ZERO_STRUCT(req);
1453
1454 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
1455 req.hdr.ctdb_magic = CTDB_MAGIC;
1456 req.hdr.ctdb_version = CTDB_PROTOCOL;
1457 req.hdr.operation = CTDB_REQ_CALL;
1458 req.hdr.reqid = ctdbd_next_reqid(conn);
1459 req.flags = flags;
1460 req.callid = CTDB_FETCH_FUNC;
1461 req.db_id = db_id;
1462 req.keylen = key.dsize;
1463
1464 iov[0].iov_base = &req;
1465 iov[0].iov_len = offsetof(struct ctdb_req_call, data);
1466 iov[1].iov_base = key.dptr;
1467 iov[1].iov_len = key.dsize;
1468
1469 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
1470 if (nwritten == -1) {
1471 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
1472 cluster_fatal("cluster dispatch daemon msg write error\n");
1473 }
1474
1475 status = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
1476
1477 if (!NT_STATUS_IS_OK(status)) {
1478 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
1479 goto fail;
1480 }
1481
1482 if (hdr->operation != CTDB_REPLY_CALL) {
1483 DEBUG(0, ("received invalid reply\n"));
1484 status = NT_STATUS_INTERNAL_ERROR;
1485 goto fail;
1486 }
1487 reply = (struct ctdb_reply_call *)hdr;
1488
1489 if (reply->datalen == 0) {
1490 /*
1491 * Treat an empty record as non-existing
1492 */
1493 status = NT_STATUS_NOT_FOUND;
1494 goto fail;
1495 }
1496
1497 parser(key, make_tdb_data(&reply->data[0], reply->datalen),
1498 private_data);
1499
1500 status = NT_STATUS_OK;
1501 fail:
1502 TALLOC_FREE(hdr);
1503 return status;
1504 }
1505
1506 /*
1507 Traverse a ctdb database. This uses a kind-of hackish way to open a second
1508 connection to ctdbd to avoid the hairy recursive and async problems with
1509 everything in-line.
1510 */
1511
1512 NTSTATUS ctdbd_traverse(uint32_t db_id,
1513 void (*fn)(TDB_DATA key, TDB_DATA data,
1514 void *private_data),
1515 void *private_data)
1516 {
1517 struct ctdbd_connection *conn;
1518 NTSTATUS status;
1519
1520 TDB_DATA key, data;
1521 struct ctdb_traverse_start t;
1522 int cstatus;
1523
1524 become_root();
1525 status = ctdbd_init_connection(NULL, &conn);
1526 unbecome_root();
1527 if (!NT_STATUS_IS_OK(status)) {
1528 DEBUG(0, ("ctdbd_init_connection failed: %s\n",
1529 nt_errstr(status)));
1530 return status;
1531 }
1532
1533 t.db_id = db_id;
1534 t.srvid = conn->rand_srvid;
1535 t.reqid = ctdbd_next_reqid(conn);
1536
1537 data.dptr = (uint8_t *)&t;
1538 data.dsize = sizeof(t);
1539
1540 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1541 CTDB_CONTROL_TRAVERSE_START, conn->rand_srvid, 0,
1542 data, NULL, NULL, &cstatus);
1543
1544 if (!NT_STATUS_IS_OK(status) || (cstatus != 0)) {
1545
1546 DEBUG(0,("ctdbd_control failed: %s, %d\n", nt_errstr(status),
1547 cstatus));
1548
1549 if (NT_STATUS_IS_OK(status)) {
1550 /*
1551 * We need a mapping here
1552 */
1553 status = NT_STATUS_UNSUCCESSFUL;
1554 }
1555 TALLOC_FREE(conn);
1556 return status;
1557 }
1558
1559 while (True) {
1560 struct ctdb_req_header *hdr;
1561 struct ctdb_req_message *m;
1562 struct ctdb_rec_data *d;
1563
1564 status = ctdb_read_packet(conn->fd, conn, &hdr);
1565 if (!NT_STATUS_IS_OK(status)) {
1566 DEBUG(0, ("ctdb_read_packet failed: %s\n",
1567 nt_errstr(status)));
1568 cluster_fatal("ctdbd died\n");
1569 }
1570
1571 if (hdr->operation != CTDB_REQ_MESSAGE) {
1572 DEBUG(0, ("Got operation %u, expected a message\n",
1573 (unsigned)hdr->operation));
1574 TALLOC_FREE(conn);
1575 return NT_STATUS_UNEXPECTED_IO_ERROR;
1576 }
1577
1578 m = (struct ctdb_req_message *)hdr;
1579 d = (struct ctdb_rec_data *)&m->data[0];
1580 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1581 DEBUG(0, ("Got invalid traverse data of length %d\n",
1582 (int)m->datalen));
1583 TALLOC_FREE(conn);
1584 return NT_STATUS_UNEXPECTED_IO_ERROR;
1585 }
1586
1587 key.dsize = d->keylen;
1588 key.dptr = &d->data[0];
1589 data.dsize = d->datalen;
1590 data.dptr = &d->data[d->keylen];
1591
1592 if (key.dsize == 0 && data.dsize == 0) {
1593 /* end of traverse */
1594 TALLOC_FREE(conn);
1595 return NT_STATUS_OK;
1596 }
1597
1598 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1599 DEBUG(0, ("Got invalid ltdb header length %d\n",
1600 (int)data.dsize));
1601 TALLOC_FREE(conn);
1602 return NT_STATUS_UNEXPECTED_IO_ERROR;
1603 }
1604 data.dsize -= sizeof(struct ctdb_ltdb_header);
1605 data.dptr += sizeof(struct ctdb_ltdb_header);
1606
1607 if (fn != NULL) {
1608 fn(key, data, private_data);
1609 }
1610 }
1611 return NT_STATUS_OK;
1612 }
1613
1614 /*
1615 This is used to canonicalize a ctdb_sock_addr structure.
1616 */
1617 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1618 struct sockaddr_storage *out)
1619 {
1620 memcpy(out, in, sizeof (*out));
1621
1622 #ifdef HAVE_IPV6
1623 if (in->ss_family == AF_INET6) {
1624 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1625 const struct sockaddr_in6 *in6 =
1626 (const struct sockaddr_in6 *)in;
1627 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1628 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1629 memset(out, 0, sizeof(*out));
1630 #ifdef HAVE_SOCK_SIN_LEN
1631 out4->sin_len = sizeof(*out);
1632 #endif
1633 out4->sin_family = AF_INET;
1634 out4->sin_port = in6->sin6_port;
1635 memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr[12], 4);
1636 }
1637 }
1638 #endif
1639 }
1640
1641 /*
1642 * Register us as a server for a particular tcp connection
1643 */
1644
1645 NTSTATUS ctdbd_register_ips(struct ctdbd_connection *conn,
1646 const struct sockaddr_storage *_server,
1647 const struct sockaddr_storage *_client,
1648 bool (*release_ip_handler)(const char *ip_addr,
1649 void *private_data),
1650 void *private_data)
1651 {
1652 /*
1653 * we still use ctdb_control_tcp for ipv4
1654 * because we want to work against older ctdb
1655 * versions at runtime
1656 */
1657 struct ctdb_control_tcp p4;
1658 struct ctdb_control_tcp_addr p;
1659 TDB_DATA data;
1660 NTSTATUS status;
1661 struct sockaddr_storage client;
1662 struct sockaddr_storage server;
1663
1664 /*
1665 * Only one connection so far
1666 */
1667 SMB_ASSERT(conn->release_ip_handler == NULL);
1668
1669 smbd_ctdb_canonicalize_ip(_client, &client);
1670 smbd_ctdb_canonicalize_ip(_server, &server);
1671
1672 switch (client.ss_family) {
1673 case AF_INET:
1674 memcpy(&p4.dest, &server, sizeof(p4.dest));
1675 memcpy(&p4.src, &client, sizeof(p4.src));
1676 data.dptr = (uint8_t *)&p4;
1677 data.dsize = sizeof(p4);
1678 break;
1679 case AF_INET6:
1680 memcpy(&p.dest.ip6, &server, sizeof(p.dest.ip6));
1681 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1682 data.dptr = (uint8_t *)&p;
1683 data.dsize = sizeof(p);
1684 break;
1685 default:
1686 return NT_STATUS_INTERNAL_ERROR;
1687 }
1688
1689 conn->release_ip_handler = release_ip_handler;
1690 conn->release_ip_priv = private_data;
1691
1692 /*
1693 * We want to be told about IP releases
1694 */
1695
1696 status = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP);
1697 if (!NT_STATUS_IS_OK(status)) {
1698 return status;
1699 }
1700
1701 /*
1702 * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1703 * can send an extra ack to trigger a reset for our client, so it
1704 * immediately reconnects
1705 */
1706 return ctdbd_control(conn, CTDB_CURRENT_NODE,
1707 CTDB_CONTROL_TCP_CLIENT, 0,
1708 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL, NULL);
1709 }
1710
1711 /*
1712 * We want to handle reconfigure events
1713 */
1714 NTSTATUS ctdbd_register_reconfigure(struct ctdbd_connection *conn)
1715 {
1716 return register_with_ctdbd(conn, CTDB_SRVID_RECONFIGURE);
1717 }
1718
1719 /*
1720 call a control on the local node
1721 */
1722 NTSTATUS ctdbd_control_local(struct ctdbd_connection *conn, uint32_t opcode,
1723 uint64_t srvid, uint32_t flags, TDB_DATA data,
1724 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1725 int *cstatus)
1726 {
1727 return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data, mem_ctx, outdata, cstatus);
1728 }
1729
1730 NTSTATUS ctdb_watch_us(struct ctdbd_connection *conn)
1731 {
1732 struct ctdb_client_notify_register reg_data;
1733 size_t struct_len;
1734 NTSTATUS status;
1735 int cstatus;
1736
1737 reg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1738 reg_data.len = 1;
1739 reg_data.notify_data[0] = 0;
1740
1741 struct_len = offsetof(struct ctdb_client_notify_register,
1742 notify_data) + reg_data.len;
1743
1744 status = ctdbd_control_local(
1745 conn, CTDB_CONTROL_REGISTER_NOTIFY, conn->rand_srvid, 0,
1746 make_tdb_data((uint8_t *)&reg_data, struct_len),
1747 NULL, NULL, &cstatus);
1748 if (!NT_STATUS_IS_OK(status)) {
1749 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1750 nt_errstr(status)));
1751 }
1752 return status;
1753 }
1754
1755 NTSTATUS ctdb_unwatch(struct ctdbd_connection *conn)
1756 {
1757 struct ctdb_client_notify_deregister dereg_data;
1758 NTSTATUS status;
1759 int cstatus;
1760
1761 dereg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1762
1763 status = ctdbd_control_local(
1764 conn, CTDB_CONTROL_DEREGISTER_NOTIFY, conn->rand_srvid, 0,
1765 make_tdb_data((uint8_t *)&dereg_data, sizeof(dereg_data)),
1766 NULL, NULL, &cstatus);
1767 if (!NT_STATUS_IS_OK(status)) {
1768 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1769 nt_errstr(status)));
1770 }
1771 return status;
1772 }
1773
1774 NTSTATUS ctdbd_probe(void)
1775 {
1776 /*
1777 * Do a very early check if ctdbd is around to avoid an abort and core
1778 * later
1779 */
1780 struct ctdbd_connection *conn = NULL;
1781 NTSTATUS status;
1782
1783 status = ctdbd_messaging_connection(talloc_tos(), &conn);
1784
1785 /*
1786 * We only care if we can connect.
1787 */
1788 TALLOC_FREE(conn);
1789
1790 return status;
1791 }
Samba GIT mirror