?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
s4-smbtorture: use struct torture_context in RPC-LSA.
[Samba.git] / source3 / lib / ctdbd_conn.c
blobdde377581b93ed601ab97b45c6a78642dc96fb79
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(NULL, 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
547 /* only messages to our pid or the broadcast are valid here */
548 if (msg->srvid != sys_getpid() && msg->srvid != MSG_SRVID_SAMBA) {
549 DEBUG(0,("Got unexpected message with srvid=%llu\n",
550 (unsigned long long)msg->srvid));
551 TALLOC_FREE(buf);
552 return NT_STATUS_OK;
553 }
554
555 if (!(msg_rec = ctdb_pull_messaging_rec(NULL, length, msg))) {
556 DEBUG(10, ("ctdb_pull_messaging_rec failed\n"));
557 TALLOC_FREE(buf);
558 return NT_STATUS_NO_MEMORY;
559 }
560
561 messaging_dispatch_rec(conn->msg_ctx, msg_rec);
562
563 TALLOC_FREE(msg_rec);
564 TALLOC_FREE(buf);
565 return NT_STATUS_OK;
566 }
567
568 /*
569 * The ctdbd socket is readable asynchronuously
570 */
571
572 static void ctdbd_socket_handler(struct event_context *event_ctx,
573 struct fd_event *event,
574 uint16 flags,
575 void *private_data)
576 {
577 struct ctdbd_connection *conn = talloc_get_type_abort(
578 private_data, struct ctdbd_connection);
579
580 NTSTATUS status;
581
582 status = packet_fd_read(conn->pkt);
583
584 if (!NT_STATUS_IS_OK(status)) {
585 DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status)));
586 cluster_fatal("ctdbd died\n");
587 }
588
589 while (packet_handler(conn->pkt, ctdb_req_complete,
590 ctdb_handle_message, conn, &status)) {
591 if (!NT_STATUS_IS_OK(status)) {
592 DEBUG(10, ("could not handle incoming message: %s\n",
593 nt_errstr(status)));
594 }
595 }
596 }
597
598 /*
599 * Prepare a ctdbd connection to receive messages
600 */
601
602 NTSTATUS ctdbd_register_msg_ctx(struct ctdbd_connection *conn,
603 struct messaging_context *msg_ctx)
604 {
605 SMB_ASSERT(conn->msg_ctx == NULL);
606 SMB_ASSERT(conn->fde == NULL);
607
608 if (!(conn->fde = event_add_fd(msg_ctx->event_ctx, conn,
609 packet_get_fd(conn->pkt),
610 EVENT_FD_READ,
611 ctdbd_socket_handler,
612 conn))) {
613 DEBUG(0, ("event_add_fd failed\n"));
614 return NT_STATUS_NO_MEMORY;
615 }
616
617 conn->msg_ctx = msg_ctx;
618
619 return NT_STATUS_OK;
620 }
621
622 /*
623 * Send a messaging message across a ctdbd
624 */
625
626 NTSTATUS ctdbd_messaging_send(struct ctdbd_connection *conn,
627 uint32 dst_vnn, uint64 dst_srvid,
628 struct messaging_rec *msg)
629 {
630 struct ctdb_req_message r;
631 TALLOC_CTX *mem_ctx;
632 DATA_BLOB blob;
633 NTSTATUS status;
634 enum ndr_err_code ndr_err;
635
636 if (!(mem_ctx = talloc_init("ctdbd_messaging_send"))) {
637 DEBUG(0, ("talloc failed\n"));
638 return NT_STATUS_NO_MEMORY;
639 }
640
641 ndr_err = ndr_push_struct_blob(
642 &blob, mem_ctx, NULL, msg,
643 (ndr_push_flags_fn_t)ndr_push_messaging_rec);
644
645 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
646 DEBUG(0, ("ndr_push_struct_blob failed: %s\n",
647 ndr_errstr(ndr_err)));
648 status = ndr_map_error2ntstatus(ndr_err);
649 goto fail;
650 }
651
652 r.hdr.length = offsetof(struct ctdb_req_message, data) + blob.length;
653 r.hdr.ctdb_magic = CTDB_MAGIC;
654 r.hdr.ctdb_version = CTDB_VERSION;
655 r.hdr.generation = 1;
656 r.hdr.operation = CTDB_REQ_MESSAGE;
657 r.hdr.destnode = dst_vnn;
658 r.hdr.srcnode = conn->our_vnn;
659 r.hdr.reqid = 0;
660 r.srvid = dst_srvid;
661 r.datalen = blob.length;
662
663 DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
664 ctdb_packet_dump(&r.hdr);
665
666 status = packet_send(
667 conn->pkt, 2,
668 data_blob_const(&r, offsetof(struct ctdb_req_message, data)),
669 blob);
670
671 if (!NT_STATUS_IS_OK(status)) {
672 DEBUG(0, ("packet_send failed: %s\n", nt_errstr(status)));
673 goto fail;
674 }
675
676 status = packet_flush(conn->pkt);
677
678 if (!NT_STATUS_IS_OK(status)) {
679 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
680 cluster_fatal("cluster dispatch daemon msg write error\n");
681 }
682
683 status = NT_STATUS_OK;
684 fail:
685 TALLOC_FREE(mem_ctx);
686 return status;
687 }
688
689 /*
690 * send/recv a generic ctdb control message
691 */
692 static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
693 uint32_t vnn, uint32 opcode,
694 uint64_t srvid, uint32_t flags,
695 TDB_DATA data,
696 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
697 int *cstatus)
698 {
699 struct ctdb_req_control req;
700 struct ctdb_reply_control *reply = NULL;
701 struct ctdbd_connection *new_conn = NULL;
702 NTSTATUS status;
703
704 /* the samba3 ctdb code can't handle NOREPLY yet */
705 flags &= ~CTDB_CTRL_FLAG_NOREPLY;
706
707 if (conn == NULL) {
708 status = ctdbd_init_connection(NULL, &new_conn);
709
710 if (!NT_STATUS_IS_OK(status)) {
711 DEBUG(10, ("Could not init temp connection: %s\n",
712 nt_errstr(status)));
713 goto fail;
714 }
715
716 conn = new_conn;
717 }
718
719 ZERO_STRUCT(req);
720 req.hdr.length = offsetof(struct ctdb_req_control, data) + data.dsize;
721 req.hdr.ctdb_magic = CTDB_MAGIC;
722 req.hdr.ctdb_version = CTDB_VERSION;
723 req.hdr.operation = CTDB_REQ_CONTROL;
724 req.hdr.reqid = ++conn->reqid;
725 req.hdr.destnode = vnn;
726 req.opcode = opcode;
727 req.srvid = srvid;
728 req.datalen = data.dsize;
729
730 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
731 ctdb_packet_dump(&req.hdr);
732
733 status = packet_send(
734 conn->pkt, 2,
735 data_blob_const(&req, offsetof(struct ctdb_req_control, data)),
736 data_blob_const(data.dptr, data.dsize));
737
738 if (!NT_STATUS_IS_OK(status)) {
739 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
740 goto fail;
741 }
742
743 status = packet_flush(conn->pkt);
744
745 if (!NT_STATUS_IS_OK(status)) {
746 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
747 cluster_fatal("cluster dispatch daemon control write error\n");
748 }
749
750 if (flags & CTDB_CTRL_FLAG_NOREPLY) {
751 TALLOC_FREE(new_conn);
752 return NT_STATUS_OK;
753 }
754
755 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
756
757 if (!NT_STATUS_IS_OK(status)) {
758 DEBUG(10, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
759 goto fail;
760 }
761
762 if (reply->hdr.operation != CTDB_REPLY_CONTROL) {
763 DEBUG(0, ("received invalid reply\n"));
764 goto fail;
765 }
766
767 if (outdata) {
768 if (!(outdata->dptr = (uint8 *)talloc_memdup(
769 mem_ctx, reply->data, reply->datalen))) {
770 TALLOC_FREE(reply);
771 return NT_STATUS_NO_MEMORY;
772 }
773 outdata->dsize = reply->datalen;
774 }
775 if (cstatus) {
776 (*cstatus) = reply->status;
777 }
778
779 status = NT_STATUS_OK;
780
781 fail:
782 TALLOC_FREE(new_conn);
783 TALLOC_FREE(reply);
784 return status;
785 }
786
787 /*
788 * see if a remote process exists
789 */
790 bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32 vnn, pid_t pid)
791 {
792 NTSTATUS status;
793 TDB_DATA data;
794 int32_t cstatus;
795
796 data.dptr = (uint8_t*)&pid;
797 data.dsize = sizeof(pid);
798
799 status = ctdbd_control(conn, vnn, CTDB_CONTROL_PROCESS_EXISTS, 0, 0,
800 data, NULL, NULL, &cstatus);
801 if (!NT_STATUS_IS_OK(status)) {
802 DEBUG(0, (__location__ " ctdb_control for process_exists "
803 "failed\n"));
804 return False;
805 }
806
807 return cstatus == 0;
808 }
809
810 /*
811 * Get a db path
812 */
813 char *ctdbd_dbpath(struct ctdbd_connection *conn,
814 TALLOC_CTX *mem_ctx, uint32_t db_id)
815 {
816 NTSTATUS status;
817 TDB_DATA data;
818 int32_t cstatus;
819
820 data.dptr = (uint8_t*)&db_id;
821 data.dsize = sizeof(db_id);
822
823 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
824 CTDB_CONTROL_GETDBPATH, 0, 0, data,
825 mem_ctx, &data, &cstatus);
826 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
827 DEBUG(0,(__location__ " ctdb_control for getdbpath failed\n"));
828 return NULL;
829 }
830
831 return (char *)data.dptr;
832 }
833
834 /*
835 * attach to a ctdb database
836 */
837 NTSTATUS ctdbd_db_attach(struct ctdbd_connection *conn,
838 const char *name, uint32_t *db_id, int tdb_flags)
839 {
840 NTSTATUS status;
841 TDB_DATA data;
842 int32_t cstatus;
843 bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
844
845 data.dptr = (uint8_t*)name;
846 data.dsize = strlen(name)+1;
847
848 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
849 persistent
850 ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
851 : CTDB_CONTROL_DB_ATTACH,
852 0, 0, data, NULL, &data, &cstatus);
853 if (!NT_STATUS_IS_OK(status)) {
854 DEBUG(0, (__location__ " ctdb_control for db_attach "
855 "failed: %s\n", nt_errstr(status)));
856 return status;
857 }
858
859 if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
860 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
861 return NT_STATUS_INTERNAL_ERROR;
862 }
863
864 *db_id = *(uint32_t *)data.dptr;
865 talloc_free(data.dptr);
866
867 if (!(tdb_flags & TDB_SEQNUM)) {
868 return NT_STATUS_OK;
869 }
870
871 data.dptr = (uint8_t *)db_id;
872 data.dsize = sizeof(*db_id);
873
874 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
875 CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
876 NULL, NULL, &cstatus);
877 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
878 DEBUG(0,(__location__ " ctdb_control for enable seqnum "
879 "failed\n"));
880 return NT_STATUS_IS_OK(status) ? NT_STATUS_INTERNAL_ERROR :
881 status;
882 }
883
884 return NT_STATUS_OK;
885 }
886
887 /*
888 * force the migration of a record to this node
889 */
890 NTSTATUS ctdbd_migrate(struct ctdbd_connection *conn, uint32 db_id,
891 TDB_DATA key)
892 {
893 struct ctdb_req_call req;
894 struct ctdb_reply_call *reply;
895 NTSTATUS status;
896
897 ZERO_STRUCT(req);
898
899 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
900 req.hdr.ctdb_magic = CTDB_MAGIC;
901 req.hdr.ctdb_version = CTDB_VERSION;
902 req.hdr.operation = CTDB_REQ_CALL;
903 req.hdr.reqid = ++conn->reqid;
904 req.flags = CTDB_IMMEDIATE_MIGRATION;
905 req.callid = CTDB_NULL_FUNC;
906 req.db_id = db_id;
907 req.keylen = key.dsize;
908
909 DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
910 ctdb_packet_dump(&req.hdr);
911
912 status = packet_send(
913 conn->pkt, 2,
914 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
915 data_blob_const(key.dptr, key.dsize));
916
917 if (!NT_STATUS_IS_OK(status)) {
918 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
919 return status;
920 }
921
922 status = packet_flush(conn->pkt);
923
924 if (!NT_STATUS_IS_OK(status)) {
925 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
926 cluster_fatal("cluster dispatch daemon control write error\n");
927 }
928
929 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
930
931 if (!NT_STATUS_IS_OK(status)) {
932 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
933 goto fail;
934 }
935
936 if (reply->hdr.operation != CTDB_REPLY_CALL) {
937 DEBUG(0, ("received invalid reply\n"));
938 status = NT_STATUS_INTERNAL_ERROR;
939 goto fail;
940 }
941
942 status = NT_STATUS_OK;
943 fail:
944
945 TALLOC_FREE(reply);
946 return status;
947 }
948
949 /*
950 * remotely fetch a record without locking it or forcing a migration
951 */
952 NTSTATUS ctdbd_fetch(struct ctdbd_connection *conn, uint32 db_id,
953 TDB_DATA key, TALLOC_CTX *mem_ctx, TDB_DATA *data)
954 {
955 struct ctdb_req_call req;
956 struct ctdb_reply_call *reply;
957 NTSTATUS status;
958
959 ZERO_STRUCT(req);
960
961 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
962 req.hdr.ctdb_magic = CTDB_MAGIC;
963 req.hdr.ctdb_version = CTDB_VERSION;
964 req.hdr.operation = CTDB_REQ_CALL;
965 req.hdr.reqid = ++conn->reqid;
966 req.flags = 0;
967 req.callid = CTDB_FETCH_FUNC;
968 req.db_id = db_id;
969 req.keylen = key.dsize;
970
971 status = packet_send(
972 conn->pkt, 2,
973 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
974 data_blob_const(key.dptr, key.dsize));
975
976 if (!NT_STATUS_IS_OK(status)) {
977 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
978 return status;
979 }
980
981 status = packet_flush(conn->pkt);
982
983 if (!NT_STATUS_IS_OK(status)) {
984 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
985 cluster_fatal("cluster dispatch daemon control write error\n");
986 }
987
988 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
989
990 if (!NT_STATUS_IS_OK(status)) {
991 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
992 goto fail;
993 }
994
995 if (reply->hdr.operation != CTDB_REPLY_CALL) {
996 DEBUG(0, ("received invalid reply\n"));
997 status = NT_STATUS_INTERNAL_ERROR;
998 goto fail;
999 }
1000
1001 data->dsize = reply->datalen;
1002 if (data->dsize == 0) {
1003 data->dptr = NULL;
1004 goto done;
1005 }
1006
1007 data->dptr = (uint8 *)talloc_memdup(mem_ctx, &reply->data[0],
1008 reply->datalen);
1009 if (data->dptr == NULL) {
1010 DEBUG(0, ("talloc failed\n"));
1011 status = NT_STATUS_NO_MEMORY;
1012 goto fail;
1013 }
1014
1015 done:
1016 status = NT_STATUS_OK;
1017 fail:
1018 TALLOC_FREE(reply);
1019 return status;
1020 }
1021
1022 struct ctdbd_traverse_state {
1023 void (*fn)(TDB_DATA key, TDB_DATA data, void *private_data);
1024 void *private_data;
1025 };
1026
1027 /*
1028 * Handle a traverse record coming in on the ctdbd connection
1029 */
1030
1031 static NTSTATUS ctdb_traverse_handler(uint8_t *buf, size_t length,
1032 void *private_data)
1033 {
1034 struct ctdbd_traverse_state *state =
1035 (struct ctdbd_traverse_state *)private_data;
1036
1037 struct ctdb_req_message *m;
1038 struct ctdb_rec_data *d;
1039 TDB_DATA key, data;
1040
1041 m = (struct ctdb_req_message *)buf;
1042
1043 if (length < sizeof(*m) || m->hdr.length != length) {
1044 DEBUG(0, ("Got invalid message of length %d\n", (int)length));
1045 TALLOC_FREE(buf);
1046 return NT_STATUS_UNEXPECTED_IO_ERROR;
1047 }
1048
1049 d = (struct ctdb_rec_data *)&m->data[0];
1050 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1051 DEBUG(0, ("Got invalid traverse data of length %d\n",
1052 (int)m->datalen));
1053 TALLOC_FREE(buf);
1054 return NT_STATUS_UNEXPECTED_IO_ERROR;
1055 }
1056
1057 key.dsize = d->keylen;
1058 key.dptr = &d->data[0];
1059 data.dsize = d->datalen;
1060 data.dptr = &d->data[d->keylen];
1061
1062 if (key.dsize == 0 && data.dsize == 0) {
1063 /* end of traverse */
1064 return NT_STATUS_END_OF_FILE;
1065 }
1066
1067 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1068 DEBUG(0, ("Got invalid ltdb header length %d\n",
1069 (int)data.dsize));
1070 TALLOC_FREE(buf);
1071 return NT_STATUS_UNEXPECTED_IO_ERROR;
1072 }
1073 data.dsize -= sizeof(struct ctdb_ltdb_header);
1074 data.dptr += sizeof(struct ctdb_ltdb_header);
1075
1076 if (state->fn) {
1077 state->fn(key, data, state->private_data);
1078 }
1079
1080 TALLOC_FREE(buf);
1081 return NT_STATUS_OK;
1082 }
1083
1084 /*
1085 Traverse a ctdb database. This uses a kind-of hackish way to open a second
1086 connection to ctdbd to avoid the hairy recursive and async problems with
1087 everything in-line.
1088 */
1089
1090 NTSTATUS ctdbd_traverse(uint32 db_id,
1091 void (*fn)(TDB_DATA key, TDB_DATA data,
1092 void *private_data),
1093 void *private_data)
1094 {
1095 struct ctdbd_connection *conn;
1096 NTSTATUS status;
1097
1098 TDB_DATA data;
1099 struct ctdb_traverse_start t;
1100 int cstatus;
1101 struct ctdbd_traverse_state state;
1102
1103 status = ctdbd_init_connection(NULL, &conn);
1104 if (!NT_STATUS_IS_OK(status)) {
1105 DEBUG(0, ("ctdbd_init_connection failed: %s\n",
1106 nt_errstr(status)));
1107 return status;
1108 }
1109
1110 t.db_id = db_id;
1111 t.srvid = conn->rand_srvid;
1112 t.reqid = ++conn->reqid;
1113
1114 data.dptr = (uint8_t *)&t;
1115 data.dsize = sizeof(t);
1116
1117 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1118 CTDB_CONTROL_TRAVERSE_START, conn->rand_srvid, 0,
1119 data, NULL, NULL, &cstatus);
1120
1121 if (!NT_STATUS_IS_OK(status) || (cstatus != 0)) {
1122
1123 DEBUG(0,("ctdbd_control failed: %s, %d\n", nt_errstr(status),
1124 cstatus));
1125
1126 if (NT_STATUS_IS_OK(status)) {
1127 /*
1128 * We need a mapping here
1129 */
1130 status = NT_STATUS_UNSUCCESSFUL;
1131 }
1132 goto done;
1133 }
1134
1135 state.fn = fn;
1136 state.private_data = private_data;
1137
1138 while (True) {
1139
1140 status = NT_STATUS_OK;
1141
1142 if (packet_handler(conn->pkt, ctdb_req_complete,
1143 ctdb_traverse_handler, &state, &status)) {
1144
1145 if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1146 status = NT_STATUS_OK;
1147 break;
1148 }
1149
1150 /*
1151 * There might be more in the queue
1152 */
1153 continue;
1154 }
1155
1156 if (!NT_STATUS_IS_OK(status)) {
1157 break;
1158 }
1159
1160 status = packet_fd_read_sync(conn->pkt);
1161
1162 if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
1163 /*
1164 * There might be more in the queue
1165 */
1166 continue;
1167 }
1168
1169 if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1170 status = NT_STATUS_OK;
1171 }
1172
1173 if (!NT_STATUS_IS_OK(status)) {
1174 DEBUG(0, ("packet_fd_read_sync failed: %s\n", nt_errstr(status)));
1175 cluster_fatal("ctdbd died\n");
1176 }
1177 }
1178
1179 done:
1180 TALLOC_FREE(conn);
1181 return status;
1182 }
1183
1184 /*
1185 This is used to canonicalize a ctdb_sock_addr structure.
1186 */
1187 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1188 struct sockaddr_storage *out)
1189 {
1190 memcpy(out, in, sizeof (*out));
1191
1192 #ifdef HAVE_IPV6
1193 if (in->ss_family == AF_INET6) {
1194 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1195 const struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)in;
1196 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1197 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1198 memset(out, 0, sizeof(*out));
1199 #ifdef HAVE_SOCK_SIN_LEN
1200 out4->sin_len = sizeof(*out);
1201 #endif
1202 out4->sin_family = AF_INET;
1203 out4->sin_port = in6->sin6_port;
1204 memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr32[3], 4);
1205 }
1206 }
1207 #endif
1208 }
1209
1210 /*
1211 * Register us as a server for a particular tcp connection
1212 */
1213
1214 NTSTATUS ctdbd_register_ips(struct ctdbd_connection *conn,
1215 const struct sockaddr_storage *_server,
1216 const struct sockaddr_storage *_client,
1217 void (*release_ip_handler)(const char *ip_addr,
1218 void *private_data),
1219 void *private_data)
1220 {
1221 /*
1222 * we still use ctdb_control_tcp for ipv4
1223 * because we want to work against older ctdb
1224 * versions at runtime
1225 */
1226 struct ctdb_control_tcp p4;
1227 #ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1228 struct ctdb_control_tcp_addr p;
1229 #endif
1230 TDB_DATA data;
1231 NTSTATUS status;
1232 struct sockaddr_storage client;
1233 struct sockaddr_storage server;
1234
1235 /*
1236 * Only one connection so far
1237 */
1238 SMB_ASSERT(conn->release_ip_handler == NULL);
1239
1240 smbd_ctdb_canonicalize_ip(_client, &client);
1241 smbd_ctdb_canonicalize_ip(_server, &server);
1242
1243 switch (client.ss_family) {
1244 case AF_INET:
1245 p4.dest = *(struct sockaddr_in *)&server;
1246 p4.src = *(struct sockaddr_in *)&client;
1247 data.dptr = (uint8_t *)&p4;
1248 data.dsize = sizeof(p4);
1249 break;
1250 #ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1251 case AF_INET6:
1252 p.dest.ip6 = *(struct sockaddr_in6 *)&server;
1253 p.src.ip6 = *(struct sockaddr_in6 *)&client;
1254 data.dptr = (uint8_t *)&p;
1255 data.dsize = sizeof(p);
1256 break;
1257 #endif
1258 default:
1259 return NT_STATUS_INTERNAL_ERROR;
1260 }
1261
1262 conn->release_ip_handler = release_ip_handler;
1263
1264 /*
1265 * We want to be told about IP releases
1266 */
1267
1268 status = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP);
1269 if (!NT_STATUS_IS_OK(status)) {
1270 return status;
1271 }
1272
1273 /*
1274 * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1275 * can send an extra ack to trigger a reset for our client, so it
1276 * immediately reconnects
1277 */
1278 return ctdbd_control(conn, CTDB_CURRENT_NODE,
1279 CTDB_CONTROL_TCP_CLIENT, 0,
1280 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL, NULL);
1281 }
1282
1283 /*
1284 * We want to handle reconfigure events
1285 */
1286 NTSTATUS ctdbd_register_reconfigure(struct ctdbd_connection *conn)
1287 {
1288 return register_with_ctdbd(conn, CTDB_SRVID_RECONFIGURE);
1289 }
1290
1291 /*
1292 call a control on the local node
1293 */
1294 NTSTATUS ctdbd_control_local(struct ctdbd_connection *conn, uint32 opcode,
1295 uint64_t srvid, uint32_t flags, TDB_DATA data,
1296 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1297 int *cstatus)
1298 {
1299 return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data, mem_ctx, outdata, cstatus);
1300 }
1301
1302 #else
1303
1304 NTSTATUS ctdbd_init_connection(TALLOC_CTX *mem_ctx,
1305 struct ctdbd_connection **pconn)
1306 {
1307 return NT_STATUS_NOT_IMPLEMENTED;
1308 }
1309
1310 #endif
Samba GIT mirror