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