?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Fix some nonempty blank lines
[Samba/aatanasov.git] / source3 / lib / ctdbd_conn.c
blob0af420412b5ec566ceb57bbf5310c1a753e1f5a5
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
23 #ifdef CLUSTER_SUPPORT
24
25 #include "librpc/gen_ndr/messaging.h"
26 #include "librpc/gen_ndr/ndr_messaging.h"
27
28 /* paths to these include files come from --with-ctdb= in configure */
29 #include "ctdb.h"
30 #include "ctdb_private.h"
31
32 struct ctdbd_connection {
33 struct messaging_context *msg_ctx;
34 uint32 reqid;
35 uint32 our_vnn;
36 uint64 rand_srvid;
37 struct packet_context *pkt;
38 struct fd_event *fde;
39
40 void (*release_ip_handler)(const char *ip_addr, void *private_data);
41 void *release_ip_priv;
42 };
43
44 static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
45 uint32_t vnn, uint32 opcode,
46 uint64_t srvid, uint32_t flags, TDB_DATA data,
47 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
48 int *cstatus);
49
50 /*
51 * exit on fatal communications errors with the ctdbd daemon
52 */
53 static void cluster_fatal(const char *why)
54 {
55 DEBUG(0,("cluster fatal event: %s - exiting immediately\n", why));
56 /* we don't use smb_panic() as we don't want to delay to write
57 a core file. We need to release this process id immediately
58 so that someone else can take over without getting sharing
59 violations */
60 _exit(0);
61 }
62
63 /*
64 *
65 */
66 static void ctdb_packet_dump(struct ctdb_req_header *hdr)
67 {
68 if (DEBUGLEVEL < 10) {
69 return;
70 }
71 DEBUGADD(10, ("len=%d, magic=%x, vers=%d, gen=%d, op=%d, reqid=%d\n",
72 (int)hdr->length, (int)hdr->ctdb_magic,
73 (int)hdr->ctdb_version, (int)hdr->generation,
74 (int)hdr->operation, (int)hdr->reqid));
75 }
76
77 /*
78 * Register a srvid with ctdbd
79 */
80 static NTSTATUS register_with_ctdbd(struct ctdbd_connection *conn,
81 uint64_t srvid)
82 {
83
84 int cstatus;
85 return ctdbd_control(conn, CTDB_CURRENT_NODE,
86 CTDB_CONTROL_REGISTER_SRVID, srvid, 0,
87 tdb_null, NULL, NULL, &cstatus);
88 }
89
90 /*
91 * get our vnn from the cluster
92 */
93 static NTSTATUS get_cluster_vnn(struct ctdbd_connection *conn, uint32 *vnn)
94 {
95 int32_t cstatus=-1;
96 NTSTATUS status;
97 status = ctdbd_control(conn,
98 CTDB_CURRENT_NODE, CTDB_CONTROL_GET_PNN, 0, 0,
99 tdb_null, NULL, NULL, &cstatus);
100 if (!NT_STATUS_IS_OK(status)) {
101 cluster_fatal("ctdbd_control failed\n");
102 }
103 *vnn = (uint32_t)cstatus;
104 return status;
105 }
106
107 uint32 ctdbd_vnn(const struct ctdbd_connection *conn)
108 {
109 return conn->our_vnn;
110 }
111
112 /*
113 * Get us a ctdb connection
114 */
115
116 static NTSTATUS ctdbd_connect(TALLOC_CTX *mem_ctx,
117 struct packet_context **presult)
118 {
119 struct packet_context *result;
120 const char *sockname = lp_ctdbd_socket();
121 struct sockaddr_un addr;
122 int fd;
123
124 if (!sockname || !*sockname) {
125 sockname = CTDB_PATH;
126 }
127
128 fd = socket(AF_UNIX, SOCK_STREAM, 0);
129 if (fd == -1) {
130 DEBUG(3, ("Could not create socket: %s\n", strerror(errno)));
131 return map_nt_error_from_unix(errno);
132 }
133
134 ZERO_STRUCT(addr);
135 addr.sun_family = AF_UNIX;
136 strncpy(addr.sun_path, sockname, sizeof(addr.sun_path));
137
138 if (sys_connect(fd, (struct sockaddr *)&addr) == -1) {
139 DEBUG(1, ("connect(%s) failed: %s\n", sockname,
140 strerror(errno)));
141 close(fd);
142 return map_nt_error_from_unix(errno);
143 }
144
145 if (!(result = packet_init(mem_ctx, fd))) {
146 close(fd);
147 return NT_STATUS_NO_MEMORY;
148 }
149
150 *presult = result;
151 return NT_STATUS_OK;
152 }
153
154 /*
155 * Do we have a complete ctdb packet in the queue?
156 */
157
158 static bool ctdb_req_complete(const uint8_t *buf, size_t available,
159 size_t *length,
160 void *private_data)
161 {
162 uint32 msglen;
163
164 if (available < sizeof(msglen)) {
165 return False;
166 }
167
168 msglen = *((uint32 *)buf);
169
170 DEBUG(10, ("msglen = %d\n", msglen));
171
172 if (msglen < sizeof(struct ctdb_req_header)) {
173 DEBUG(0, ("Got invalid msglen: %d, expected at least %d for "
174 "the req_header\n", (int)msglen,
175 (int)sizeof(struct ctdb_req_header)));
176 cluster_fatal("ctdbd protocol error\n");
177 }
178
179 if (available < msglen) {
180 return false;
181 }
182
183 *length = msglen;
184 return true;
185 }
186
187 /*
188 * State necessary to defer an incoming message while we are waiting for a
189 * ctdb reply.
190 */
191
192 struct deferred_msg_state {
193 struct messaging_context *msg_ctx;
194 struct messaging_rec *rec;
195 };
196
197 /*
198 * Timed event handler for the deferred message
199 */
200
201 static void deferred_message_dispatch(struct event_context *event_ctx,
202 struct timed_event *te,
203 struct timeval now,
204 void *private_data)
205 {
206 struct deferred_msg_state *state = talloc_get_type_abort(
207 private_data, struct deferred_msg_state);
208
209 messaging_dispatch_rec(state->msg_ctx, state->rec);
210 TALLOC_FREE(state);
211 TALLOC_FREE(te);
212 }
213
214 struct req_pull_state {
215 TALLOC_CTX *mem_ctx;
216 DATA_BLOB req;
217 };
218
219 /*
220 * Pull a ctdb request out of the incoming packet queue
221 */
222
223 static NTSTATUS ctdb_req_pull(uint8_t *buf, size_t length,
224 void *private_data)
225 {
226 struct req_pull_state *state = (struct req_pull_state *)private_data;
227
228 state->req.data = talloc_move(state->mem_ctx, &buf);
229 state->req.length = length;
230 return NT_STATUS_OK;
231 }
232
233 /*
234 * Fetch a messaging_rec from an incoming ctdb style message
235 */
236
237 static struct messaging_rec *ctdb_pull_messaging_rec(TALLOC_CTX *mem_ctx,
238 size_t overall_length,
239 struct ctdb_req_message *msg)
240 {
241 struct messaging_rec *result;
242 DATA_BLOB blob;
243 enum ndr_err_code ndr_err;
244
245 if ((overall_length < offsetof(struct ctdb_req_message, data))
246 || (overall_length
247 < offsetof(struct ctdb_req_message, data) + msg->datalen)) {
248
249 cluster_fatal("got invalid msg length");
250 }
251
252 if (!(result = TALLOC_P(mem_ctx, struct messaging_rec))) {
253 DEBUG(0, ("talloc failed\n"));
254 return NULL;
255 }
256
257 blob = data_blob_const(msg->data, msg->datalen);
258
259 ndr_err = ndr_pull_struct_blob(
260 &blob, result, NULL, result,
261 (ndr_pull_flags_fn_t)ndr_pull_messaging_rec);
262
263 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
264 DEBUG(0, ("ndr_pull_struct_blob failed: %s\n",
265 ndr_errstr(ndr_err)));
266 TALLOC_FREE(result);
267 return NULL;
268 }
269
270 if (DEBUGLEVEL >= 10) {
271 DEBUG(10, ("ctdb_pull_messaging_rec:\n"));
272 NDR_PRINT_DEBUG(messaging_rec, result);
273 }
274
275 return result;
276 }
277
278 /*
279 * Read a full ctdbd request. If we have a messaging context, defer incoming
280 * messages that might come in between.
281 */
282
283 static NTSTATUS ctdb_read_req(struct ctdbd_connection *conn, uint32 reqid,
284 TALLOC_CTX *mem_ctx, void *result)
285 {
286 struct ctdb_req_header *hdr;
287 struct req_pull_state state;
288 NTSTATUS status;
289
290 again:
291
292 status = packet_fd_read_sync(conn->pkt);
293
294 if (NT_STATUS_EQUAL(status, NT_STATUS_NETWORK_BUSY)) {
295 /* EAGAIN */
296 goto again;
297 } else if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
298 /* EAGAIN */
299 goto again;
300 }
301
302 if (!NT_STATUS_IS_OK(status)) {
303 DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status)));
304 cluster_fatal("ctdbd died\n");
305 }
306
307 next_pkt:
308
309 ZERO_STRUCT(state);
310 state.mem_ctx = mem_ctx;
311
312 if (!packet_handler(conn->pkt, ctdb_req_complete, ctdb_req_pull,
313 &state, &status)) {
314 /*
315 * Not enough data
316 */
317 DEBUG(10, ("not enough data from ctdb socket, retrying\n"));
318 goto again;
319 }
320
321 if (!NT_STATUS_IS_OK(status)) {
322 DEBUG(0, ("Could not read packet: %s\n", nt_errstr(status)));
323 cluster_fatal("ctdbd died\n");
324 }
325
326 hdr = (struct ctdb_req_header *)state.req.data;
327
328 DEBUG(10, ("Received ctdb packet\n"));
329 ctdb_packet_dump(hdr);
330
331 if (hdr->operation == CTDB_REQ_MESSAGE) {
332 struct timed_event *evt;
333 struct deferred_msg_state *msg_state;
334 struct ctdb_req_message *msg = (struct ctdb_req_message *)hdr;
335
336 if (conn->msg_ctx == NULL) {
337 DEBUG(1, ("Got a message without having a msg ctx, "
338 "dropping msg %llu\n",
339 (long long unsigned)msg->srvid));
340 goto next_pkt;
341 }
342
343 if ((conn->release_ip_handler != NULL)
344 && (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
345 /* must be dispatched immediately */
346 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
347 conn->release_ip_handler((const char *)msg->data,
348 conn->release_ip_priv);
349 TALLOC_FREE(hdr);
350 goto next_pkt;
351 }
352
353 if (msg->srvid == CTDB_SRVID_RECONFIGURE) {
354 DEBUG(0,("Got cluster reconfigure message in ctdb_read_req\n"));
355 messaging_send(conn->msg_ctx, procid_self(),
356 MSG_SMB_BRL_VALIDATE, &data_blob_null);
357 TALLOC_FREE(hdr);
358 goto next_pkt;
359 }
360
361 if (!(msg_state = TALLOC_P(talloc_autofree_context(), struct deferred_msg_state))) {
362 DEBUG(0, ("talloc failed\n"));
363 TALLOC_FREE(hdr);
364 goto next_pkt;
365 }
366
367 if (!(msg_state->rec = ctdb_pull_messaging_rec(
368 msg_state, state.req.length, msg))) {
369 DEBUG(0, ("ctdbd_pull_messaging_rec failed\n"));
370 TALLOC_FREE(msg_state);
371 TALLOC_FREE(hdr);
372 goto next_pkt;
373 }
374
375 TALLOC_FREE(hdr);
376
377 msg_state->msg_ctx = conn->msg_ctx;
378
379 /*
380 * We're waiting for a call reply, but an async message has
381 * crossed. Defer dispatching to the toplevel event loop.
382 */
383 evt = event_add_timed(conn->msg_ctx->event_ctx,
384 conn->msg_ctx->event_ctx,
385 timeval_zero(),
386 deferred_message_dispatch,
387 msg_state);
388 if (evt == NULL) {
389 DEBUG(0, ("event_add_timed failed\n"));
390 TALLOC_FREE(msg_state);
391 TALLOC_FREE(hdr);
392 goto next_pkt;
393 }
394
395 goto next_pkt;
396 }
397
398 if (hdr->reqid != reqid) {
399 /* we got the wrong reply */
400 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
401 "been %u\n", hdr->reqid, reqid));
402 TALLOC_FREE(hdr);
403 goto again;
404 }
405
406 *((void **)result) = talloc_move(mem_ctx, &hdr);
407
408 return NT_STATUS_OK;
409 }
410
411 /*
412 * Get us a ctdbd connection
413 */
414
415 NTSTATUS ctdbd_init_connection(TALLOC_CTX *mem_ctx,
416 struct ctdbd_connection **pconn)
417 {
418 struct ctdbd_connection *conn;
419 NTSTATUS status;
420
421 if (!(conn = TALLOC_ZERO_P(mem_ctx, struct ctdbd_connection))) {
422 DEBUG(0, ("talloc failed\n"));
423 return NT_STATUS_NO_MEMORY;
424 }
425
426 status = ctdbd_connect(conn, &conn->pkt);
427
428 if (!NT_STATUS_IS_OK(status)) {
429 DEBUG(10, ("ctdbd_connect failed: %s\n", nt_errstr(status)));
430 goto fail;
431 }
432
433 status = get_cluster_vnn(conn, &conn->our_vnn);
434
435 if (!NT_STATUS_IS_OK(status)) {
436 DEBUG(10, ("get_cluster_vnn failed: %s\n", nt_errstr(status)));
437 goto fail;
438 }
439
440 generate_random_buffer((unsigned char *)&conn->rand_srvid,
441 sizeof(conn->rand_srvid));
442
443 status = register_with_ctdbd(conn, conn->rand_srvid);
444
445 if (!NT_STATUS_IS_OK(status)) {
446 DEBUG(5, ("Could not register random srvid: %s\n",
447 nt_errstr(status)));
448 goto fail;
449 }
450
451 *pconn = conn;
452 return NT_STATUS_OK;
453
454 fail:
455 TALLOC_FREE(conn);
456 return status;
457 }
458
459 /*
460 * Get us a ctdbd connection and register us as a process
461 */
462
463 NTSTATUS ctdbd_messaging_connection(TALLOC_CTX *mem_ctx,
464 struct ctdbd_connection **pconn)
465 {
466 struct ctdbd_connection *conn;
467 NTSTATUS status;
468
469 status = ctdbd_init_connection(mem_ctx, &conn);
470
471 if (!NT_STATUS_IS_OK(status)) {
472 return status;
473 }
474
475 status = register_with_ctdbd(conn, (uint64_t)sys_getpid());
476 if (!NT_STATUS_IS_OK(status)) {
477 goto fail;
478 }
479
480 status = register_with_ctdbd(conn, MSG_SRVID_SAMBA);
481 if (!NT_STATUS_IS_OK(status)) {
482 goto fail;
483 }
484
485 *pconn = conn;
486 return NT_STATUS_OK;
487
488 fail:
489 TALLOC_FREE(conn);
490 return status;
491 }
492
493 /*
494 * Packet handler to receive and handle a ctdb message
495 */
496 static NTSTATUS ctdb_handle_message(uint8_t *buf, size_t length,
497 void *private_data)
498 {
499 struct ctdbd_connection *conn = talloc_get_type_abort(
500 private_data, struct ctdbd_connection);
501 struct ctdb_req_message *msg;
502 struct messaging_rec *msg_rec;
503
504 msg = (struct ctdb_req_message *)buf;
505
506 if (msg->hdr.operation != CTDB_REQ_MESSAGE) {
507 DEBUG(0, ("Received async msg of type %u, discarding\n",
508 msg->hdr.operation));
509 TALLOC_FREE(buf);
510 return NT_STATUS_INVALID_PARAMETER;
511 }
512
513 if ((conn->release_ip_handler != NULL)
514 && (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
515 /* must be dispatched immediately */
516 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
517 conn->release_ip_handler((const char *)msg->data,
518 conn->release_ip_priv);
519 TALLOC_FREE(buf);
520 return NT_STATUS_OK;
521 }
522
523 SMB_ASSERT(conn->msg_ctx != NULL);
524
525 if (msg->srvid == CTDB_SRVID_RECONFIGURE) {
526 DEBUG(0,("Got cluster reconfigure message\n"));
527 /*
528 * when the cluster is reconfigured, we need to clean the brl
529 * database
530 */
531 messaging_send(conn->msg_ctx, procid_self(),
532 MSG_SMB_BRL_VALIDATE, &data_blob_null);
533
534 /*
535 * it's possible that we have just rejoined the cluster after
536 * an outage. In that case our pending locks could have been
537 * removed from the lockdb, so retry them once more
538 */
539 message_send_all(conn->msg_ctx, MSG_SMB_UNLOCK, NULL, 0, NULL);
540
541 TALLOC_FREE(buf);
542
543 return NT_STATUS_OK;
544 }
545
546 /* only messages to our pid or the broadcast are valid here */
547 if (msg->srvid != sys_getpid() && msg->srvid != MSG_SRVID_SAMBA) {
548 DEBUG(0,("Got unexpected message with srvid=%llu\n",
549 (unsigned long long)msg->srvid));
550 TALLOC_FREE(buf);
551 return NT_STATUS_OK;
552 }
553
554 if (!(msg_rec = ctdb_pull_messaging_rec(NULL, length, msg))) {
555 DEBUG(10, ("ctdb_pull_messaging_rec failed\n"));
556 TALLOC_FREE(buf);
557 return NT_STATUS_NO_MEMORY;
558 }
559
560 messaging_dispatch_rec(conn->msg_ctx, msg_rec);
561
562 TALLOC_FREE(msg_rec);
563 TALLOC_FREE(buf);
564 return NT_STATUS_OK;
565 }
566
567 /*
568 * The ctdbd socket is readable asynchronuously
569 */
570
571 static void ctdbd_socket_handler(struct event_context *event_ctx,
572 struct fd_event *event,
573 uint16 flags,
574 void *private_data)
575 {
576 struct ctdbd_connection *conn = talloc_get_type_abort(
577 private_data, struct ctdbd_connection);
578
579 NTSTATUS status;
580
581 status = packet_fd_read(conn->pkt);
582
583 if (!NT_STATUS_IS_OK(status)) {
584 DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status)));
585 cluster_fatal("ctdbd died\n");
586 }
587
588 while (packet_handler(conn->pkt, ctdb_req_complete,
589 ctdb_handle_message, conn, &status)) {
590 if (!NT_STATUS_IS_OK(status)) {
591 DEBUG(10, ("could not handle incoming message: %s\n",
592 nt_errstr(status)));
593 }
594 }
595 }
596
597 /*
598 * Prepare a ctdbd connection to receive messages
599 */
600
601 NTSTATUS ctdbd_register_msg_ctx(struct ctdbd_connection *conn,
602 struct messaging_context *msg_ctx)
603 {
604 SMB_ASSERT(conn->msg_ctx == NULL);
605 SMB_ASSERT(conn->fde == NULL);
606
607 if (!(conn->fde = event_add_fd(msg_ctx->event_ctx, conn,
608 packet_get_fd(conn->pkt),
609 EVENT_FD_READ,
610 ctdbd_socket_handler,
611 conn))) {
612 DEBUG(0, ("event_add_fd failed\n"));
613 return NT_STATUS_NO_MEMORY;
614 }
615
616 conn->msg_ctx = msg_ctx;
617
618 return NT_STATUS_OK;
619 }
620
621 /*
622 * Send a messaging message across a ctdbd
623 */
624
625 NTSTATUS ctdbd_messaging_send(struct ctdbd_connection *conn,
626 uint32 dst_vnn, uint64 dst_srvid,
627 struct messaging_rec *msg)
628 {
629 struct ctdb_req_message r;
630 TALLOC_CTX *mem_ctx;
631 DATA_BLOB blob;
632 NTSTATUS status;
633 enum ndr_err_code ndr_err;
634
635 if (!(mem_ctx = talloc_init("ctdbd_messaging_send"))) {
636 DEBUG(0, ("talloc failed\n"));
637 return NT_STATUS_NO_MEMORY;
638 }
639
640 ndr_err = ndr_push_struct_blob(
641 &blob, mem_ctx, NULL, msg,
642 (ndr_push_flags_fn_t)ndr_push_messaging_rec);
643
644 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
645 DEBUG(0, ("ndr_push_struct_blob failed: %s\n",
646 ndr_errstr(ndr_err)));
647 status = ndr_map_error2ntstatus(ndr_err);
648 goto fail;
649 }
650
651 r.hdr.length = offsetof(struct ctdb_req_message, data) + blob.length;
652 r.hdr.ctdb_magic = CTDB_MAGIC;
653 r.hdr.ctdb_version = CTDB_VERSION;
654 r.hdr.generation = 1;
655 r.hdr.operation = CTDB_REQ_MESSAGE;
656 r.hdr.destnode = dst_vnn;
657 r.hdr.srcnode = conn->our_vnn;
658 r.hdr.reqid = 0;
659 r.srvid = dst_srvid;
660 r.datalen = blob.length;
661
662 DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
663 ctdb_packet_dump(&r.hdr);
664
665 status = packet_send(
666 conn->pkt, 2,
667 data_blob_const(&r, offsetof(struct ctdb_req_message, data)),
668 blob);
669
670 if (!NT_STATUS_IS_OK(status)) {
671 DEBUG(0, ("packet_send failed: %s\n", nt_errstr(status)));
672 goto fail;
673 }
674
675 status = packet_flush(conn->pkt);
676
677 if (!NT_STATUS_IS_OK(status)) {
678 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
679 cluster_fatal("cluster dispatch daemon msg write error\n");
680 }
681
682 status = NT_STATUS_OK;
683 fail:
684 TALLOC_FREE(mem_ctx);
685 return status;
686 }
687
688 /*
689 * send/recv a generic ctdb control message
690 */
691 static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
692 uint32_t vnn, uint32 opcode,
693 uint64_t srvid, uint32_t flags,
694 TDB_DATA data,
695 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
696 int *cstatus)
697 {
698 struct ctdb_req_control req;
699 struct ctdb_reply_control *reply = NULL;
700 struct ctdbd_connection *new_conn = NULL;
701 NTSTATUS status;
702
703 /* the samba3 ctdb code can't handle NOREPLY yet */
704 flags &= ~CTDB_CTRL_FLAG_NOREPLY;
705
706 if (conn == NULL) {
707 status = ctdbd_init_connection(NULL, &new_conn);
708
709 if (!NT_STATUS_IS_OK(status)) {
710 DEBUG(10, ("Could not init temp connection: %s\n",
711 nt_errstr(status)));
712 goto fail;
713 }
714
715 conn = new_conn;
716 }
717
718 ZERO_STRUCT(req);
719 req.hdr.length = offsetof(struct ctdb_req_control, data) + data.dsize;
720 req.hdr.ctdb_magic = CTDB_MAGIC;
721 req.hdr.ctdb_version = CTDB_VERSION;
722 req.hdr.operation = CTDB_REQ_CONTROL;
723 req.hdr.reqid = ++conn->reqid;
724 req.hdr.destnode = vnn;
725 req.opcode = opcode;
726 req.srvid = srvid;
727 req.datalen = data.dsize;
728
729 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
730 ctdb_packet_dump(&req.hdr);
731
732 status = packet_send(
733 conn->pkt, 2,
734 data_blob_const(&req, offsetof(struct ctdb_req_control, data)),
735 data_blob_const(data.dptr, data.dsize));
736
737 if (!NT_STATUS_IS_OK(status)) {
738 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
739 goto fail;
740 }
741
742 status = packet_flush(conn->pkt);
743
744 if (!NT_STATUS_IS_OK(status)) {
745 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
746 cluster_fatal("cluster dispatch daemon control write error\n");
747 }
748
749 if (flags & CTDB_CTRL_FLAG_NOREPLY) {
750 TALLOC_FREE(new_conn);
751 return NT_STATUS_OK;
752 }
753
754 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
755
756 if (!NT_STATUS_IS_OK(status)) {
757 DEBUG(10, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
758 goto fail;
759 }
760
761 if (reply->hdr.operation != CTDB_REPLY_CONTROL) {
762 DEBUG(0, ("received invalid reply\n"));
763 goto fail;
764 }
765
766 if (outdata) {
767 if (!(outdata->dptr = (uint8 *)talloc_memdup(
768 mem_ctx, reply->data, reply->datalen))) {
769 TALLOC_FREE(reply);
770 return NT_STATUS_NO_MEMORY;
771 }
772 outdata->dsize = reply->datalen;
773 }
774 if (cstatus) {
775 (*cstatus) = reply->status;
776 }
777
778 status = NT_STATUS_OK;
779
780 fail:
781 TALLOC_FREE(new_conn);
782 TALLOC_FREE(reply);
783 return status;
784 }
785
786 /*
787 * see if a remote process exists
788 */
789 bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32 vnn, pid_t pid)
790 {
791 NTSTATUS status;
792 TDB_DATA data;
793 int32_t cstatus;
794
795 data.dptr = (uint8_t*)&pid;
796 data.dsize = sizeof(pid);
797
798 status = ctdbd_control(conn, vnn, CTDB_CONTROL_PROCESS_EXISTS, 0, 0,
799 data, NULL, NULL, &cstatus);
800 if (!NT_STATUS_IS_OK(status)) {
801 DEBUG(0, (__location__ " ctdb_control for process_exists "
802 "failed\n"));
803 return False;
804 }
805
806 return cstatus == 0;
807 }
808
809 /*
810 * Get a db path
811 */
812 char *ctdbd_dbpath(struct ctdbd_connection *conn,
813 TALLOC_CTX *mem_ctx, uint32_t db_id)
814 {
815 NTSTATUS status;
816 TDB_DATA data;
817 int32_t cstatus;
818
819 data.dptr = (uint8_t*)&db_id;
820 data.dsize = sizeof(db_id);
821
822 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
823 CTDB_CONTROL_GETDBPATH, 0, 0, data,
824 mem_ctx, &data, &cstatus);
825 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
826 DEBUG(0,(__location__ " ctdb_control for getdbpath failed\n"));
827 return NULL;
828 }
829
830 return (char *)data.dptr;
831 }
832
833 /*
834 * attach to a ctdb database
835 */
836 NTSTATUS ctdbd_db_attach(struct ctdbd_connection *conn,
837 const char *name, uint32_t *db_id, int tdb_flags)
838 {
839 NTSTATUS status;
840 TDB_DATA data;
841 int32_t cstatus;
842 bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
843
844 data.dptr = (uint8_t*)name;
845 data.dsize = strlen(name)+1;
846
847 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
848 persistent
849 ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
850 : CTDB_CONTROL_DB_ATTACH,
851 0, 0, data, NULL, &data, &cstatus);
852 if (!NT_STATUS_IS_OK(status)) {
853 DEBUG(0, (__location__ " ctdb_control for db_attach "
854 "failed: %s\n", nt_errstr(status)));
855 return status;
856 }
857
858 if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
859 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
860 return NT_STATUS_INTERNAL_ERROR;
861 }
862
863 *db_id = *(uint32_t *)data.dptr;
864 talloc_free(data.dptr);
865
866 if (!(tdb_flags & TDB_SEQNUM)) {
867 return NT_STATUS_OK;
868 }
869
870 data.dptr = (uint8_t *)db_id;
871 data.dsize = sizeof(*db_id);
872
873 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
874 CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
875 NULL, NULL, &cstatus);
876 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
877 DEBUG(0,(__location__ " ctdb_control for enable seqnum "
878 "failed\n"));
879 return NT_STATUS_IS_OK(status) ? NT_STATUS_INTERNAL_ERROR :
880 status;
881 }
882
883 return NT_STATUS_OK;
884 }
885
886 /*
887 * force the migration of a record to this node
888 */
889 NTSTATUS ctdbd_migrate(struct ctdbd_connection *conn, uint32 db_id,
890 TDB_DATA key)
891 {
892 struct ctdb_req_call req;
893 struct ctdb_reply_call *reply;
894 NTSTATUS status;
895
896 ZERO_STRUCT(req);
897
898 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
899 req.hdr.ctdb_magic = CTDB_MAGIC;
900 req.hdr.ctdb_version = CTDB_VERSION;
901 req.hdr.operation = CTDB_REQ_CALL;
902 req.hdr.reqid = ++conn->reqid;
903 req.flags = CTDB_IMMEDIATE_MIGRATION;
904 req.callid = CTDB_NULL_FUNC;
905 req.db_id = db_id;
906 req.keylen = key.dsize;
907
908 DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
909 ctdb_packet_dump(&req.hdr);
910
911 status = packet_send(
912 conn->pkt, 2,
913 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
914 data_blob_const(key.dptr, key.dsize));
915
916 if (!NT_STATUS_IS_OK(status)) {
917 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
918 return status;
919 }
920
921 status = packet_flush(conn->pkt);
922
923 if (!NT_STATUS_IS_OK(status)) {
924 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
925 cluster_fatal("cluster dispatch daemon control write error\n");
926 }
927
928 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
929
930 if (!NT_STATUS_IS_OK(status)) {
931 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
932 goto fail;
933 }
934
935 if (reply->hdr.operation != CTDB_REPLY_CALL) {
936 DEBUG(0, ("received invalid reply\n"));
937 status = NT_STATUS_INTERNAL_ERROR;
938 goto fail;
939 }
940
941 status = NT_STATUS_OK;
942 fail:
943
944 TALLOC_FREE(reply);
945 return status;
946 }
947
948 /*
949 * remotely fetch a record without locking it or forcing a migration
950 */
951 NTSTATUS ctdbd_fetch(struct ctdbd_connection *conn, uint32 db_id,
952 TDB_DATA key, TALLOC_CTX *mem_ctx, TDB_DATA *data)
953 {
954 struct ctdb_req_call req;
955 struct ctdb_reply_call *reply;
956 NTSTATUS status;
957
958 ZERO_STRUCT(req);
959
960 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
961 req.hdr.ctdb_magic = CTDB_MAGIC;
962 req.hdr.ctdb_version = CTDB_VERSION;
963 req.hdr.operation = CTDB_REQ_CALL;
964 req.hdr.reqid = ++conn->reqid;
965 req.flags = 0;
966 req.callid = CTDB_FETCH_FUNC;
967 req.db_id = db_id;
968 req.keylen = key.dsize;
969
970 status = packet_send(
971 conn->pkt, 2,
972 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
973 data_blob_const(key.dptr, key.dsize));
974
975 if (!NT_STATUS_IS_OK(status)) {
976 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
977 return status;
978 }
979
980 status = packet_flush(conn->pkt);
981
982 if (!NT_STATUS_IS_OK(status)) {
983 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
984 cluster_fatal("cluster dispatch daemon control write error\n");
985 }
986
987 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
988
989 if (!NT_STATUS_IS_OK(status)) {
990 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
991 goto fail;
992 }
993
994 if (reply->hdr.operation != CTDB_REPLY_CALL) {
995 DEBUG(0, ("received invalid reply\n"));
996 status = NT_STATUS_INTERNAL_ERROR;
997 goto fail;
998 }
999
1000 data->dsize = reply->datalen;
1001 if (data->dsize == 0) {
1002 data->dptr = NULL;
1003 goto done;
1004 }
1005
1006 data->dptr = (uint8 *)talloc_memdup(mem_ctx, &reply->data[0],
1007 reply->datalen);
1008 if (data->dptr == NULL) {
1009 DEBUG(0, ("talloc failed\n"));
1010 status = NT_STATUS_NO_MEMORY;
1011 goto fail;
1012 }
1013
1014 done:
1015 status = NT_STATUS_OK;
1016 fail:
1017 TALLOC_FREE(reply);
1018 return status;
1019 }
1020
1021 struct ctdbd_traverse_state {
1022 void (*fn)(TDB_DATA key, TDB_DATA data, void *private_data);
1023 void *private_data;
1024 };
1025
1026 /*
1027 * Handle a traverse record coming in on the ctdbd connection
1028 */
1029
1030 static NTSTATUS ctdb_traverse_handler(uint8_t *buf, size_t length,
1031 void *private_data)
1032 {
1033 struct ctdbd_traverse_state *state =
1034 (struct ctdbd_traverse_state *)private_data;
1035
1036 struct ctdb_req_message *m;
1037 struct ctdb_rec_data *d;
1038 TDB_DATA key, data;
1039
1040 m = (struct ctdb_req_message *)buf;
1041
1042 if (length < sizeof(*m) || m->hdr.length != length) {
1043 DEBUG(0, ("Got invalid message of length %d\n", (int)length));
1044 TALLOC_FREE(buf);
1045 return NT_STATUS_UNEXPECTED_IO_ERROR;
1046 }
1047
1048 d = (struct ctdb_rec_data *)&m->data[0];
1049 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1050 DEBUG(0, ("Got invalid traverse data of length %d\n",
1051 (int)m->datalen));
1052 TALLOC_FREE(buf);
1053 return NT_STATUS_UNEXPECTED_IO_ERROR;
1054 }
1055
1056 key.dsize = d->keylen;
1057 key.dptr = &d->data[0];
1058 data.dsize = d->datalen;
1059 data.dptr = &d->data[d->keylen];
1060
1061 if (key.dsize == 0 && data.dsize == 0) {
1062 /* end of traverse */
1063 return NT_STATUS_END_OF_FILE;
1064 }
1065
1066 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1067 DEBUG(0, ("Got invalid ltdb header length %d\n",
1068 (int)data.dsize));
1069 TALLOC_FREE(buf);
1070 return NT_STATUS_UNEXPECTED_IO_ERROR;
1071 }
1072 data.dsize -= sizeof(struct ctdb_ltdb_header);
1073 data.dptr += sizeof(struct ctdb_ltdb_header);
1074
1075 if (state->fn) {
1076 state->fn(key, data, state->private_data);
1077 }
1078
1079 TALLOC_FREE(buf);
1080 return NT_STATUS_OK;
1081 }
1082
1083 /*
1084 Traverse a ctdb database. This uses a kind-of hackish way to open a second
1085 connection to ctdbd to avoid the hairy recursive and async problems with
1086 everything in-line.
1087 */
1088
1089 NTSTATUS ctdbd_traverse(uint32 db_id,
1090 void (*fn)(TDB_DATA key, TDB_DATA data,
1091 void *private_data),
1092 void *private_data)
1093 {
1094 struct ctdbd_connection *conn;
1095 NTSTATUS status;
1096
1097 TDB_DATA data;
1098 struct ctdb_traverse_start t;
1099 int cstatus;
1100 struct ctdbd_traverse_state state;
1101
1102 status = ctdbd_init_connection(NULL, &conn);
1103 if (!NT_STATUS_IS_OK(status)) {
1104 DEBUG(0, ("ctdbd_init_connection failed: %s\n",
1105 nt_errstr(status)));
1106 return status;
1107 }
1108
1109 t.db_id = db_id;
1110 t.srvid = conn->rand_srvid;
1111 t.reqid = ++conn->reqid;
1112
1113 data.dptr = (uint8_t *)&t;
1114 data.dsize = sizeof(t);
1115
1116 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1117 CTDB_CONTROL_TRAVERSE_START, conn->rand_srvid, 0,
1118 data, NULL, NULL, &cstatus);
1119
1120 if (!NT_STATUS_IS_OK(status) || (cstatus != 0)) {
1121
1122 DEBUG(0,("ctdbd_control failed: %s, %d\n", nt_errstr(status),
1123 cstatus));
1124
1125 if (NT_STATUS_IS_OK(status)) {
1126 /*
1127 * We need a mapping here
1128 */
1129 status = NT_STATUS_UNSUCCESSFUL;
1130 }
1131 goto done;
1132 }
1133
1134 state.fn = fn;
1135 state.private_data = private_data;
1136
1137 while (True) {
1138
1139 status = NT_STATUS_OK;
1140
1141 if (packet_handler(conn->pkt, ctdb_req_complete,
1142 ctdb_traverse_handler, &state, &status)) {
1143
1144 if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1145 status = NT_STATUS_OK;
1146 break;
1147 }
1148
1149 /*
1150 * There might be more in the queue
1151 */
1152 continue;
1153 }
1154
1155 if (!NT_STATUS_IS_OK(status)) {
1156 break;
1157 }
1158
1159 status = packet_fd_read_sync(conn->pkt);
1160
1161 if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
1162 /*
1163 * There might be more in the queue
1164 */
1165 continue;
1166 }
1167
1168 if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1169 status = NT_STATUS_OK;
1170 }
1171
1172 if (!NT_STATUS_IS_OK(status)) {
1173 DEBUG(0, ("packet_fd_read_sync failed: %s\n", nt_errstr(status)));
1174 cluster_fatal("ctdbd died\n");
1175 }
1176 }
1177
1178 done:
1179 TALLOC_FREE(conn);
1180 return status;
1181 }
1182
1183 /*
1184 This is used to canonicalize a ctdb_sock_addr structure.
1185 */
1186 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1187 struct sockaddr_storage *out)
1188 {
1189 memcpy(out, in, sizeof (*out));
1190
1191 #ifdef HAVE_IPV6
1192 if (in->ss_family == AF_INET6) {
1193 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1194 const struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)in;
1195 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1196 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1197 memset(out, 0, sizeof(*out));
1198 #ifdef HAVE_SOCK_SIN_LEN
1199 out4->sin_len = sizeof(*out);
1200 #endif
1201 out4->sin_family = AF_INET;
1202 out4->sin_port = in6->sin6_port;
1203 memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr32[3], 4);
1204 }
1205 }
1206 #endif
1207 }
1208
1209 /*
1210 * Register us as a server for a particular tcp connection
1211 */
1212
1213 NTSTATUS ctdbd_register_ips(struct ctdbd_connection *conn,
1214 const struct sockaddr_storage *_server,
1215 const struct sockaddr_storage *_client,
1216 void (*release_ip_handler)(const char *ip_addr,
1217 void *private_data),
1218 void *private_data)
1219 {
1220 /*
1221 * we still use ctdb_control_tcp for ipv4
1222 * because we want to work against older ctdb
1223 * versions at runtime
1224 */
1225 struct ctdb_control_tcp p4;
1226 #ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1227 struct ctdb_control_tcp_addr p;
1228 #endif
1229 TDB_DATA data;
1230 NTSTATUS status;
1231 struct sockaddr_storage client;
1232 struct sockaddr_storage server;
1233
1234 /*
1235 * Only one connection so far
1236 */
1237 SMB_ASSERT(conn->release_ip_handler == NULL);
1238
1239 smbd_ctdb_canonicalize_ip(_client, &client);
1240 smbd_ctdb_canonicalize_ip(_server, &server);
1241
1242 switch (client.ss_family) {
1243 case AF_INET:
1244 p4.dest = *(struct sockaddr_in *)&server;
1245 p4.src = *(struct sockaddr_in *)&client;
1246 data.dptr = (uint8_t *)&p4;
1247 data.dsize = sizeof(p4);
1248 break;
1249 #ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1250 case AF_INET6:
1251 p.dest.ip6 = *(struct sockaddr_in6 *)&server;
1252 p.src.ip6 = *(struct sockaddr_in6 *)&client;
1253 data.dptr = (uint8_t *)&p;
1254 data.dsize = sizeof(p);
1255 break;
1256 #endif
1257 default:
1258 return NT_STATUS_INTERNAL_ERROR;
1259 }
1260
1261 conn->release_ip_handler = release_ip_handler;
1262
1263 /*
1264 * We want to be told about IP releases
1265 */
1266
1267 status = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP);
1268 if (!NT_STATUS_IS_OK(status)) {
1269 return status;
1270 }
1271
1272 /*
1273 * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1274 * can send an extra ack to trigger a reset for our client, so it
1275 * immediately reconnects
1276 */
1277 return ctdbd_control(conn, CTDB_CURRENT_NODE,
1278 CTDB_CONTROL_TCP_CLIENT, 0,
1279 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL, NULL);
1280 }
1281
1282 /*
1283 * We want to handle reconfigure events
1284 */
1285 NTSTATUS ctdbd_register_reconfigure(struct ctdbd_connection *conn)
1286 {
1287 return register_with_ctdbd(conn, CTDB_SRVID_RECONFIGURE);
1288 }
1289
1290 /*
1291 call a control on the local node
1292 */
1293 NTSTATUS ctdbd_control_local(struct ctdbd_connection *conn, uint32 opcode,
1294 uint64_t srvid, uint32_t flags, TDB_DATA data,
1295 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1296 int *cstatus)
1297 {
1298 return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data, mem_ctx, outdata, cstatus);
1299 }
1300
1301 #else
1302
1303 NTSTATUS ctdbd_init_connection(TALLOC_CTX *mem_ctx,
1304 struct ctdbd_connection **pconn)
1305 {
1306 return NT_STATUS_NOT_IMPLEMENTED;
1307 }
1308
1309 #endif
aatanasov's samba4 branch