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