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