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