source: branches/samba-3.5.x/source3/lib/ctdbd_conn.c

Last change on this file was 733, checked in by Silvan Scherrer, 13 years ago

Samba Server 3.5: update branche to 3.5.13
File size: 34.2 KB
Line 
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
32struct 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
44static 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 */
53static 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 */
66static 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 */
80static 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 */
93static 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
107uint32 ctdbd_vnn(const struct ctdbd_connection *conn)
108{
109 return conn->our_vnn;
110}
111
112/*
113 * Get us a ctdb connection
114 */
115
116static 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
158static 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
192struct 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
201static 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
214struct 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
223static 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
237static 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
278static NTSTATUS ctdb_packet_fd_read_sync(struct packet_context *ctx)
279{
280 struct timeval timeout;
281 struct timeval *ptimeout;
282
283 timeout = timeval_set(lp_ctdb_timeout(), 0);
284 ptimeout = (timeout.tv_sec != 0) ? &timeout : NULL;
285
286 return packet_fd_read_sync(ctx, ptimeout);
287}
288
289/*
290 * Read a full ctdbd request. If we have a messaging context, defer incoming
291 * messages that might come in between.
292 */
293
294static NTSTATUS ctdb_read_req(struct ctdbd_connection *conn, uint32 reqid,
295 TALLOC_CTX *mem_ctx, void *result)
296{
297 struct ctdb_req_header *hdr;
298 struct req_pull_state state;
299 NTSTATUS status;
300
301 again:
302
303 status = ctdb_packet_fd_read_sync(conn->pkt);
304
305 if (NT_STATUS_EQUAL(status, NT_STATUS_NETWORK_BUSY)) {
306 /* EAGAIN */
307 goto again;
308 } else if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
309 /* EAGAIN */
310 goto again;
311 }
312
313 if (!NT_STATUS_IS_OK(status)) {
314 DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status)));
315 cluster_fatal("ctdbd died\n");
316 }
317
318 next_pkt:
319
320 ZERO_STRUCT(state);
321 state.mem_ctx = mem_ctx;
322
323 if (!packet_handler(conn->pkt, ctdb_req_complete, ctdb_req_pull,
324 &state, &status)) {
325 /*
326 * Not enough data
327 */
328 DEBUG(10, ("not enough data from ctdb socket, retrying\n"));
329 goto again;
330 }
331
332 if (!NT_STATUS_IS_OK(status)) {
333 DEBUG(0, ("Could not read packet: %s\n", nt_errstr(status)));
334 cluster_fatal("ctdbd died\n");
335 }
336
337 hdr = (struct ctdb_req_header *)state.req.data;
338
339 DEBUG(10, ("Received ctdb packet\n"));
340 ctdb_packet_dump(hdr);
341
342 if (hdr->operation == CTDB_REQ_MESSAGE) {
343 struct timed_event *evt;
344 struct deferred_msg_state *msg_state;
345 struct ctdb_req_message *msg = (struct ctdb_req_message *)hdr;
346
347 if (conn->msg_ctx == NULL) {
348 DEBUG(1, ("Got a message without having a msg ctx, "
349 "dropping msg %llu\n",
350 (long long unsigned)msg->srvid));
351 goto next_pkt;
352 }
353
354 if ((conn->release_ip_handler != NULL)
355 && (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
356 /* must be dispatched immediately */
357 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
358 conn->release_ip_handler((const char *)msg->data,
359 conn->release_ip_priv);
360 TALLOC_FREE(hdr);
361 goto next_pkt;
362 }
363
364 if ((msg->srvid == CTDB_SRVID_RECONFIGURE)
365 || (msg->srvid == CTDB_SRVID_SAMBA_NOTIFY)) {
366
367 DEBUG(1, ("ctdb_read_req: Got %s message\n",
368 (msg->srvid == CTDB_SRVID_RECONFIGURE)
369 ? "cluster reconfigure" : "SAMBA_NOTIFY"));
370
371 messaging_send(conn->msg_ctx, procid_self(),
372 MSG_SMB_BRL_VALIDATE, &data_blob_null);
373 messaging_send(conn->msg_ctx, procid_self(),
374 MSG_DBWRAP_G_LOCK_RETRY,
375 &data_blob_null);
376 TALLOC_FREE(hdr);
377 goto next_pkt;
378 }
379
380 if (!(msg_state = TALLOC_P(talloc_autofree_context(), struct deferred_msg_state))) {
381 DEBUG(0, ("talloc failed\n"));
382 TALLOC_FREE(hdr);
383 goto next_pkt;
384 }
385
386 if (!(msg_state->rec = ctdb_pull_messaging_rec(
387 msg_state, state.req.length, msg))) {
388 DEBUG(0, ("ctdbd_pull_messaging_rec failed\n"));
389 TALLOC_FREE(msg_state);
390 TALLOC_FREE(hdr);
391 goto next_pkt;
392 }
393
394 TALLOC_FREE(hdr);
395
396 msg_state->msg_ctx = conn->msg_ctx;
397
398 /*
399 * We're waiting for a call reply, but an async message has
400 * crossed. Defer dispatching to the toplevel event loop.
401 */
402 evt = event_add_timed(conn->msg_ctx->event_ctx,
403 conn->msg_ctx->event_ctx,
404 timeval_zero(),
405 deferred_message_dispatch,
406 msg_state);
407 if (evt == NULL) {
408 DEBUG(0, ("event_add_timed failed\n"));
409 TALLOC_FREE(msg_state);
410 TALLOC_FREE(hdr);
411 goto next_pkt;
412 }
413
414 goto next_pkt;
415 }
416
417 if (hdr->reqid != reqid) {
418 /* we got the wrong reply */
419 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
420 "been %u\n", hdr->reqid, reqid));
421 TALLOC_FREE(hdr);
422 goto again;
423 }
424
425 *((void **)result) = talloc_move(mem_ctx, &hdr);
426
427 return NT_STATUS_OK;
428}
429
430/*
431 * Get us a ctdbd connection
432 */
433
434NTSTATUS ctdbd_init_connection(TALLOC_CTX *mem_ctx,
435 struct ctdbd_connection **pconn)
436{
437 struct ctdbd_connection *conn;
438 NTSTATUS status;
439
440 if (!(conn = TALLOC_ZERO_P(mem_ctx, struct ctdbd_connection))) {
441 DEBUG(0, ("talloc failed\n"));
442 return NT_STATUS_NO_MEMORY;
443 }
444
445 status = ctdbd_connect(conn, &conn->pkt);
446
447 if (!NT_STATUS_IS_OK(status)) {
448 DEBUG(10, ("ctdbd_connect failed: %s\n", nt_errstr(status)));
449 goto fail;
450 }
451
452 status = get_cluster_vnn(conn, &conn->our_vnn);
453
454 if (!NT_STATUS_IS_OK(status)) {
455 DEBUG(10, ("get_cluster_vnn failed: %s\n", nt_errstr(status)));
456 goto fail;
457 }
458
459 generate_random_buffer((unsigned char *)&conn->rand_srvid,
460 sizeof(conn->rand_srvid));
461
462 status = register_with_ctdbd(conn, conn->rand_srvid);
463
464 if (!NT_STATUS_IS_OK(status)) {
465 DEBUG(5, ("Could not register random srvid: %s\n",
466 nt_errstr(status)));
467 goto fail;
468 }
469
470 *pconn = conn;
471 return NT_STATUS_OK;
472
473 fail:
474 TALLOC_FREE(conn);
475 return status;
476}
477
478/*
479 * Get us a ctdbd connection and register us as a process
480 */
481
482NTSTATUS ctdbd_messaging_connection(TALLOC_CTX *mem_ctx,
483 struct ctdbd_connection **pconn)
484{
485 struct ctdbd_connection *conn;
486 NTSTATUS status;
487
488 status = ctdbd_init_connection(mem_ctx, &conn);
489
490 if (!NT_STATUS_IS_OK(status)) {
491 return status;
492 }
493
494 status = register_with_ctdbd(conn, (uint64_t)sys_getpid());
495 if (!NT_STATUS_IS_OK(status)) {
496 goto fail;
497 }
498
499 status = register_with_ctdbd(conn, MSG_SRVID_SAMBA);
500 if (!NT_STATUS_IS_OK(status)) {
501 goto fail;
502 }
503
504 status = register_with_ctdbd(conn, CTDB_SRVID_SAMBA_NOTIFY);
505 if (!NT_STATUS_IS_OK(status)) {
506 goto fail;
507 }
508
509 *pconn = conn;
510 return NT_STATUS_OK;
511
512 fail:
513 TALLOC_FREE(conn);
514 return status;
515}
516
517struct messaging_context *ctdb_conn_msg_ctx(struct ctdbd_connection *conn)
518{
519 return conn->msg_ctx;
520}
521
522int ctdbd_conn_get_fd(struct ctdbd_connection *conn)
523{
524 return packet_get_fd(conn->pkt);
525}
526
527/*
528 * Packet handler to receive and handle a ctdb message
529 */
530static NTSTATUS ctdb_handle_message(uint8_t *buf, size_t length,
531 void *private_data)
532{
533 struct ctdbd_connection *conn = talloc_get_type_abort(
534 private_data, struct ctdbd_connection);
535 struct ctdb_req_message *msg;
536 struct messaging_rec *msg_rec;
537
538 msg = (struct ctdb_req_message *)buf;
539
540 if (msg->hdr.operation != CTDB_REQ_MESSAGE) {
541 DEBUG(0, ("Received async msg of type %u, discarding\n",
542 msg->hdr.operation));
543 TALLOC_FREE(buf);
544 return NT_STATUS_INVALID_PARAMETER;
545 }
546
547 if ((conn->release_ip_handler != NULL)
548 && (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
549 /* must be dispatched immediately */
550 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
551 conn->release_ip_handler((const char *)msg->data,
552 conn->release_ip_priv);
553 TALLOC_FREE(buf);
554 return NT_STATUS_OK;
555 }
556
557 SMB_ASSERT(conn->msg_ctx != NULL);
558
559 if ((msg->srvid == CTDB_SRVID_RECONFIGURE)
560 || (msg->srvid == CTDB_SRVID_SAMBA_NOTIFY)){
561 DEBUG(0,("Got cluster reconfigure message\n"));
562 /*
563 * when the cluster is reconfigured or someone of the
564 * family has passed away (SAMBA_NOTIFY), we need to
565 * clean the brl database
566 */
567 messaging_send(conn->msg_ctx, procid_self(),
568 MSG_SMB_BRL_VALIDATE, &data_blob_null);
569
570 messaging_send(conn->msg_ctx, procid_self(),
571 MSG_DBWRAP_G_LOCK_RETRY,
572 &data_blob_null);
573
574 TALLOC_FREE(buf);
575 return NT_STATUS_OK;
576 }
577
578 /* only messages to our pid or the broadcast are valid here */
579 if (msg->srvid != sys_getpid() && msg->srvid != MSG_SRVID_SAMBA) {
580 DEBUG(0,("Got unexpected message with srvid=%llu\n",
581 (unsigned long long)msg->srvid));
582 TALLOC_FREE(buf);
583 return NT_STATUS_OK;
584 }
585
586 if (!(msg_rec = ctdb_pull_messaging_rec(NULL, length, msg))) {
587 DEBUG(10, ("ctdb_pull_messaging_rec failed\n"));
588 TALLOC_FREE(buf);
589 return NT_STATUS_NO_MEMORY;
590 }
591
592 messaging_dispatch_rec(conn->msg_ctx, msg_rec);
593
594 TALLOC_FREE(msg_rec);
595 TALLOC_FREE(buf);
596 return NT_STATUS_OK;
597}
598
599/*
600 * The ctdbd socket is readable asynchronuously
601 */
602
603static void ctdbd_socket_handler(struct event_context *event_ctx,
604 struct fd_event *event,
605 uint16 flags,
606 void *private_data)
607{
608 struct ctdbd_connection *conn = talloc_get_type_abort(
609 private_data, struct ctdbd_connection);
610
611 NTSTATUS status;
612
613 status = packet_fd_read(conn->pkt);
614
615 if (!NT_STATUS_IS_OK(status)) {
616 DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status)));
617 cluster_fatal("ctdbd died\n");
618 }
619
620 while (packet_handler(conn->pkt, ctdb_req_complete,
621 ctdb_handle_message, conn, &status)) {
622 if (!NT_STATUS_IS_OK(status)) {
623 DEBUG(10, ("could not handle incoming message: %s\n",
624 nt_errstr(status)));
625 }
626 }
627}
628
629/*
630 * Prepare a ctdbd connection to receive messages
631 */
632
633NTSTATUS ctdbd_register_msg_ctx(struct ctdbd_connection *conn,
634 struct messaging_context *msg_ctx)
635{
636 SMB_ASSERT(conn->msg_ctx == NULL);
637 SMB_ASSERT(conn->fde == NULL);
638
639 if (!(conn->fde = event_add_fd(msg_ctx->event_ctx, conn,
640 packet_get_fd(conn->pkt),
641 EVENT_FD_READ,
642 ctdbd_socket_handler,
643 conn))) {
644 DEBUG(0, ("event_add_fd failed\n"));
645 return NT_STATUS_NO_MEMORY;
646 }
647
648 conn->msg_ctx = msg_ctx;
649
650 return NT_STATUS_OK;
651}
652
653/*
654 * Send a messaging message across a ctdbd
655 */
656
657NTSTATUS ctdbd_messaging_send(struct ctdbd_connection *conn,
658 uint32 dst_vnn, uint64 dst_srvid,
659 struct messaging_rec *msg)
660{
661 struct ctdb_req_message r;
662 TALLOC_CTX *mem_ctx;
663 DATA_BLOB blob;
664 NTSTATUS status;
665 enum ndr_err_code ndr_err;
666
667 if (!(mem_ctx = talloc_init("ctdbd_messaging_send"))) {
668 DEBUG(0, ("talloc failed\n"));
669 return NT_STATUS_NO_MEMORY;
670 }
671
672 ndr_err = ndr_push_struct_blob(
673 &blob, mem_ctx, NULL, msg,
674 (ndr_push_flags_fn_t)ndr_push_messaging_rec);
675
676 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
677 DEBUG(0, ("ndr_push_struct_blob failed: %s\n",
678 ndr_errstr(ndr_err)));
679 status = ndr_map_error2ntstatus(ndr_err);
680 goto fail;
681 }
682
683 r.hdr.length = offsetof(struct ctdb_req_message, data) + blob.length;
684 r.hdr.ctdb_magic = CTDB_MAGIC;
685 r.hdr.ctdb_version = CTDB_VERSION;
686 r.hdr.generation = 1;
687 r.hdr.operation = CTDB_REQ_MESSAGE;
688 r.hdr.destnode = dst_vnn;
689 r.hdr.srcnode = conn->our_vnn;
690 r.hdr.reqid = 0;
691 r.srvid = dst_srvid;
692 r.datalen = blob.length;
693
694 DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
695 ctdb_packet_dump(&r.hdr);
696
697 status = packet_send(
698 conn->pkt, 2,
699 data_blob_const(&r, offsetof(struct ctdb_req_message, data)),
700 blob);
701
702 if (!NT_STATUS_IS_OK(status)) {
703 DEBUG(0, ("packet_send failed: %s\n", nt_errstr(status)));
704 goto fail;
705 }
706
707 status = packet_flush(conn->pkt);
708
709 if (!NT_STATUS_IS_OK(status)) {
710 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
711 cluster_fatal("cluster dispatch daemon msg write error\n");
712 }
713
714 status = NT_STATUS_OK;
715 fail:
716 TALLOC_FREE(mem_ctx);
717 return status;
718}
719
720/*
721 * send/recv a generic ctdb control message
722 */
723static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
724 uint32_t vnn, uint32 opcode,
725 uint64_t srvid, uint32_t flags,
726 TDB_DATA data,
727 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
728 int *cstatus)
729{
730 struct ctdb_req_control req;
731 struct ctdb_reply_control *reply = NULL;
732 struct ctdbd_connection *new_conn = NULL;
733 NTSTATUS status;
734
735 /* the samba3 ctdb code can't handle NOREPLY yet */
736 flags &= ~CTDB_CTRL_FLAG_NOREPLY;
737
738 if (conn == NULL) {
739 status = ctdbd_init_connection(NULL, &new_conn);
740
741 if (!NT_STATUS_IS_OK(status)) {
742 DEBUG(10, ("Could not init temp connection: %s\n",
743 nt_errstr(status)));
744 goto fail;
745 }
746
747 conn = new_conn;
748 }
749
750 ZERO_STRUCT(req);
751 req.hdr.length = offsetof(struct ctdb_req_control, data) + data.dsize;
752 req.hdr.ctdb_magic = CTDB_MAGIC;
753 req.hdr.ctdb_version = CTDB_VERSION;
754 req.hdr.operation = CTDB_REQ_CONTROL;
755 req.hdr.reqid = ++conn->reqid;
756 req.hdr.destnode = vnn;
757 req.opcode = opcode;
758 req.srvid = srvid;
759 req.datalen = data.dsize;
760
761 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
762 ctdb_packet_dump(&req.hdr);
763
764 status = packet_send(
765 conn->pkt, 2,
766 data_blob_const(&req, offsetof(struct ctdb_req_control, data)),
767 data_blob_const(data.dptr, data.dsize));
768
769 if (!NT_STATUS_IS_OK(status)) {
770 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
771 goto fail;
772 }
773
774 status = packet_flush(conn->pkt);
775
776 if (!NT_STATUS_IS_OK(status)) {
777 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
778 cluster_fatal("cluster dispatch daemon control write error\n");
779 }
780
781 if (flags & CTDB_CTRL_FLAG_NOREPLY) {
782 TALLOC_FREE(new_conn);
783 return NT_STATUS_OK;
784 }
785
786 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
787
788 if (!NT_STATUS_IS_OK(status)) {
789 DEBUG(10, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
790 goto fail;
791 }
792
793 if (reply->hdr.operation != CTDB_REPLY_CONTROL) {
794 DEBUG(0, ("received invalid reply\n"));
795 goto fail;
796 }
797
798 if (outdata) {
799 if (!(outdata->dptr = (uint8 *)talloc_memdup(
800 mem_ctx, reply->data, reply->datalen))) {
801 TALLOC_FREE(reply);
802 return NT_STATUS_NO_MEMORY;
803 }
804 outdata->dsize = reply->datalen;
805 }
806 if (cstatus) {
807 (*cstatus) = reply->status;
808 }
809
810 status = NT_STATUS_OK;
811
812 fail:
813 TALLOC_FREE(new_conn);
814 TALLOC_FREE(reply);
815 return status;
816}
817
818/*
819 * see if a remote process exists
820 */
821bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32 vnn, pid_t pid)
822{
823 NTSTATUS status;
824 TDB_DATA data;
825 int32_t cstatus;
826
827 data.dptr = (uint8_t*)&pid;
828 data.dsize = sizeof(pid);
829
830 status = ctdbd_control(conn, vnn, CTDB_CONTROL_PROCESS_EXISTS, 0, 0,
831 data, NULL, NULL, &cstatus);
832 if (!NT_STATUS_IS_OK(status)) {
833 DEBUG(0, (__location__ " ctdb_control for process_exists "
834 "failed\n"));
835 return False;
836 }
837
838 return cstatus == 0;
839}
840
841/*
842 * Get a db path
843 */
844char *ctdbd_dbpath(struct ctdbd_connection *conn,
845 TALLOC_CTX *mem_ctx, uint32_t db_id)
846{
847 NTSTATUS status;
848 TDB_DATA data;
849 int32_t cstatus;
850
851 data.dptr = (uint8_t*)&db_id;
852 data.dsize = sizeof(db_id);
853
854 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
855 CTDB_CONTROL_GETDBPATH, 0, 0, data,
856 mem_ctx, &data, &cstatus);
857 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
858 DEBUG(0,(__location__ " ctdb_control for getdbpath failed\n"));
859 return NULL;
860 }
861
862 return (char *)data.dptr;
863}
864
865/*
866 * attach to a ctdb database
867 */
868NTSTATUS ctdbd_db_attach(struct ctdbd_connection *conn,
869 const char *name, uint32_t *db_id, int tdb_flags)
870{
871 NTSTATUS status;
872 TDB_DATA data;
873 int32_t cstatus;
874 bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
875
876 data.dptr = (uint8_t*)name;
877 data.dsize = strlen(name)+1;
878
879 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
880 persistent
881 ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
882 : CTDB_CONTROL_DB_ATTACH,
883 0, 0, data, NULL, &data, &cstatus);
884 if (!NT_STATUS_IS_OK(status)) {
885 DEBUG(0, (__location__ " ctdb_control for db_attach "
886 "failed: %s\n", nt_errstr(status)));
887 return status;
888 }
889
890 if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
891 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
892 return NT_STATUS_INTERNAL_ERROR;
893 }
894
895 *db_id = *(uint32_t *)data.dptr;
896 talloc_free(data.dptr);
897
898 if (!(tdb_flags & TDB_SEQNUM)) {
899 return NT_STATUS_OK;
900 }
901
902 data.dptr = (uint8_t *)db_id;
903 data.dsize = sizeof(*db_id);
904
905 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
906 CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
907 NULL, NULL, &cstatus);
908 if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
909 DEBUG(0,(__location__ " ctdb_control for enable seqnum "
910 "failed\n"));
911 return NT_STATUS_IS_OK(status) ? NT_STATUS_INTERNAL_ERROR :
912 status;
913 }
914
915 return NT_STATUS_OK;
916}
917
918/*
919 * force the migration of a record to this node
920 */
921NTSTATUS ctdbd_migrate(struct ctdbd_connection *conn, uint32 db_id,
922 TDB_DATA key)
923{
924 struct ctdb_req_call req;
925 struct ctdb_reply_call *reply;
926 NTSTATUS status;
927
928 ZERO_STRUCT(req);
929
930 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
931 req.hdr.ctdb_magic = CTDB_MAGIC;
932 req.hdr.ctdb_version = CTDB_VERSION;
933 req.hdr.operation = CTDB_REQ_CALL;
934 req.hdr.reqid = ++conn->reqid;
935 req.flags = CTDB_IMMEDIATE_MIGRATION;
936 req.callid = CTDB_NULL_FUNC;
937 req.db_id = db_id;
938 req.keylen = key.dsize;
939
940 DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
941 ctdb_packet_dump(&req.hdr);
942
943 status = packet_send(
944 conn->pkt, 2,
945 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
946 data_blob_const(key.dptr, key.dsize));
947
948 if (!NT_STATUS_IS_OK(status)) {
949 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
950 return status;
951 }
952
953 status = packet_flush(conn->pkt);
954
955 if (!NT_STATUS_IS_OK(status)) {
956 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
957 cluster_fatal("cluster dispatch daemon control write error\n");
958 }
959
960 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
961
962 if (!NT_STATUS_IS_OK(status)) {
963 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
964 goto fail;
965 }
966
967 if (reply->hdr.operation != CTDB_REPLY_CALL) {
968 DEBUG(0, ("received invalid reply\n"));
969 status = NT_STATUS_INTERNAL_ERROR;
970 goto fail;
971 }
972
973 status = NT_STATUS_OK;
974 fail:
975
976 TALLOC_FREE(reply);
977 return status;
978}
979
980/*
981 * remotely fetch a record without locking it or forcing a migration
982 */
983NTSTATUS ctdbd_fetch(struct ctdbd_connection *conn, uint32 db_id,
984 TDB_DATA key, TALLOC_CTX *mem_ctx, TDB_DATA *data)
985{
986 struct ctdb_req_call req;
987 struct ctdb_reply_call *reply;
988 NTSTATUS status;
989
990 ZERO_STRUCT(req);
991
992 req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
993 req.hdr.ctdb_magic = CTDB_MAGIC;
994 req.hdr.ctdb_version = CTDB_VERSION;
995 req.hdr.operation = CTDB_REQ_CALL;
996 req.hdr.reqid = ++conn->reqid;
997 req.flags = 0;
998 req.callid = CTDB_FETCH_FUNC;
999 req.db_id = db_id;
1000 req.keylen = key.dsize;
1001
1002 status = packet_send(
1003 conn->pkt, 2,
1004 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
1005 data_blob_const(key.dptr, key.dsize));
1006
1007 if (!NT_STATUS_IS_OK(status)) {
1008 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
1009 return status;
1010 }
1011
1012 status = packet_flush(conn->pkt);
1013
1014 if (!NT_STATUS_IS_OK(status)) {
1015 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
1016 cluster_fatal("cluster dispatch daemon control write error\n");
1017 }
1018
1019 status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
1020
1021 if (!NT_STATUS_IS_OK(status)) {
1022 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
1023 goto fail;
1024 }
1025
1026 if (reply->hdr.operation != CTDB_REPLY_CALL) {
1027 DEBUG(0, ("received invalid reply\n"));
1028 status = NT_STATUS_INTERNAL_ERROR;
1029 goto fail;
1030 }
1031
1032 data->dsize = reply->datalen;
1033 if (data->dsize == 0) {
1034 data->dptr = NULL;
1035 goto done;
1036 }
1037
1038 data->dptr = (uint8 *)talloc_memdup(mem_ctx, &reply->data[0],
1039 reply->datalen);
1040 if (data->dptr == NULL) {
1041 DEBUG(0, ("talloc failed\n"));
1042 status = NT_STATUS_NO_MEMORY;
1043 goto fail;
1044 }
1045
1046 done:
1047 status = NT_STATUS_OK;
1048 fail:
1049 TALLOC_FREE(reply);
1050 return status;
1051}
1052
1053struct ctdbd_traverse_state {
1054 void (*fn)(TDB_DATA key, TDB_DATA data, void *private_data);
1055 void *private_data;
1056};
1057
1058/*
1059 * Handle a traverse record coming in on the ctdbd connection
1060 */
1061
1062static NTSTATUS ctdb_traverse_handler(uint8_t *buf, size_t length,
1063 void *private_data)
1064{
1065 struct ctdbd_traverse_state *state =
1066 (struct ctdbd_traverse_state *)private_data;
1067
1068 struct ctdb_req_message *m;
1069 struct ctdb_rec_data *d;
1070 TDB_DATA key, data;
1071
1072 m = (struct ctdb_req_message *)buf;
1073
1074 if (length < sizeof(*m) || m->hdr.length != length) {
1075 DEBUG(0, ("Got invalid message of length %d\n", (int)length));
1076 TALLOC_FREE(buf);
1077 return NT_STATUS_UNEXPECTED_IO_ERROR;
1078 }
1079
1080 d = (struct ctdb_rec_data *)&m->data[0];
1081 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1082 DEBUG(0, ("Got invalid traverse data of length %d\n",
1083 (int)m->datalen));
1084 TALLOC_FREE(buf);
1085 return NT_STATUS_UNEXPECTED_IO_ERROR;
1086 }
1087
1088 key.dsize = d->keylen;
1089 key.dptr = &d->data[0];
1090 data.dsize = d->datalen;
1091 data.dptr = &d->data[d->keylen];
1092
1093 if (key.dsize == 0 && data.dsize == 0) {
1094 /* end of traverse */
1095 return NT_STATUS_END_OF_FILE;
1096 }
1097
1098 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1099 DEBUG(0, ("Got invalid ltdb header length %d\n",
1100 (int)data.dsize));
1101 TALLOC_FREE(buf);
1102 return NT_STATUS_UNEXPECTED_IO_ERROR;
1103 }
1104 data.dsize -= sizeof(struct ctdb_ltdb_header);
1105 data.dptr += sizeof(struct ctdb_ltdb_header);
1106
1107 if (state->fn) {
1108 state->fn(key, data, state->private_data);
1109 }
1110
1111 TALLOC_FREE(buf);
1112 return NT_STATUS_OK;
1113}
1114
1115/*
1116 Traverse a ctdb database. This uses a kind-of hackish way to open a second
1117 connection to ctdbd to avoid the hairy recursive and async problems with
1118 everything in-line.
1119*/
1120
1121NTSTATUS ctdbd_traverse(uint32 db_id,
1122 void (*fn)(TDB_DATA key, TDB_DATA data,
1123 void *private_data),
1124 void *private_data)
1125{
1126 struct ctdbd_connection *conn;
1127 NTSTATUS status;
1128
1129 TDB_DATA data;
1130 struct ctdb_traverse_start t;
1131 int cstatus;
1132 struct ctdbd_traverse_state state;
1133
1134 become_root();
1135 status = ctdbd_init_connection(NULL, &conn);
1136 unbecome_root();
1137 if (!NT_STATUS_IS_OK(status)) {
1138 DEBUG(0, ("ctdbd_init_connection failed: %s\n",
1139 nt_errstr(status)));
1140 return status;
1141 }
1142
1143 t.db_id = db_id;
1144 t.srvid = conn->rand_srvid;
1145 t.reqid = ++conn->reqid;
1146
1147 data.dptr = (uint8_t *)&t;
1148 data.dsize = sizeof(t);
1149
1150 status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1151 CTDB_CONTROL_TRAVERSE_START, conn->rand_srvid, 0,
1152 data, NULL, NULL, &cstatus);
1153
1154 if (!NT_STATUS_IS_OK(status) || (cstatus != 0)) {
1155
1156 DEBUG(0,("ctdbd_control failed: %s, %d\n", nt_errstr(status),
1157 cstatus));
1158
1159 if (NT_STATUS_IS_OK(status)) {
1160 /*
1161 * We need a mapping here
1162 */
1163 status = NT_STATUS_UNSUCCESSFUL;
1164 }
1165 goto done;
1166 }
1167
1168 state.fn = fn;
1169 state.private_data = private_data;
1170
1171 while (True) {
1172
1173 status = NT_STATUS_OK;
1174
1175 if (packet_handler(conn->pkt, ctdb_req_complete,
1176 ctdb_traverse_handler, &state, &status)) {
1177
1178 if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1179 status = NT_STATUS_OK;
1180 break;
1181 }
1182
1183 /*
1184 * There might be more in the queue
1185 */
1186 continue;
1187 }
1188
1189 if (!NT_STATUS_IS_OK(status)) {
1190 break;
1191 }
1192
1193 status = ctdb_packet_fd_read_sync(conn->pkt);
1194
1195 if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
1196 /*
1197 * There might be more in the queue
1198 */
1199 continue;
1200 }
1201
1202 if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1203 status = NT_STATUS_OK;
1204 break;
1205 }
1206
1207 if (!NT_STATUS_IS_OK(status)) {
1208 DEBUG(0, ("packet_fd_read_sync failed: %s\n", nt_errstr(status)));
1209 cluster_fatal("ctdbd died\n");
1210 }
1211 }
1212
1213 done:
1214 TALLOC_FREE(conn);
1215 return status;
1216}
1217
1218/*
1219 This is used to canonicalize a ctdb_sock_addr structure.
1220*/
1221static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1222 struct sockaddr_storage *out)
1223{
1224 memcpy(out, in, sizeof (*out));
1225
1226#ifdef HAVE_IPV6
1227 if (in->ss_family == AF_INET6) {
1228 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1229 const struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)in;
1230 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1231 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1232 memset(out, 0, sizeof(*out));
1233#ifdef HAVE_SOCK_SIN_LEN
1234 out4->sin_len = sizeof(*out);
1235#endif
1236 out4->sin_family = AF_INET;
1237 out4->sin_port = in6->sin6_port;
1238 memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr32[3], 4);
1239 }
1240 }
1241#endif
1242}
1243
1244/*
1245 * Register us as a server for a particular tcp connection
1246 */
1247
1248NTSTATUS ctdbd_register_ips(struct ctdbd_connection *conn,
1249 const struct sockaddr_storage *_server,
1250 const struct sockaddr_storage *_client,
1251 void (*release_ip_handler)(const char *ip_addr,
1252 void *private_data),
1253 void *private_data)
1254{
1255 /*
1256 * we still use ctdb_control_tcp for ipv4
1257 * because we want to work against older ctdb
1258 * versions at runtime
1259 */
1260 struct ctdb_control_tcp p4;
1261#ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1262 struct ctdb_control_tcp_addr p;
1263#endif
1264 TDB_DATA data;
1265 NTSTATUS status;
1266 struct sockaddr_storage client;
1267 struct sockaddr_storage server;
1268
1269 /*
1270 * Only one connection so far
1271 */
1272 SMB_ASSERT(conn->release_ip_handler == NULL);
1273
1274 smbd_ctdb_canonicalize_ip(_client, &client);
1275 smbd_ctdb_canonicalize_ip(_server, &server);
1276
1277 switch (client.ss_family) {
1278 case AF_INET:
1279 p4.dest = *(struct sockaddr_in *)&server;
1280 p4.src = *(struct sockaddr_in *)&client;
1281 data.dptr = (uint8_t *)&p4;
1282 data.dsize = sizeof(p4);
1283 break;
1284#ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1285 case AF_INET6:
1286 p.dest.ip6 = *(struct sockaddr_in6 *)&server;
1287 p.src.ip6 = *(struct sockaddr_in6 *)&client;
1288 data.dptr = (uint8_t *)&p;
1289 data.dsize = sizeof(p);
1290 break;
1291#endif
1292 default:
1293 return NT_STATUS_INTERNAL_ERROR;
1294 }
1295
1296 conn->release_ip_handler = release_ip_handler;
1297
1298 /*
1299 * We want to be told about IP releases
1300 */
1301
1302 status = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP);
1303 if (!NT_STATUS_IS_OK(status)) {
1304 return status;
1305 }
1306
1307 /*
1308 * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1309 * can send an extra ack to trigger a reset for our client, so it
1310 * immediately reconnects
1311 */
1312 return ctdbd_control(conn, CTDB_CURRENT_NODE,
1313 CTDB_CONTROL_TCP_CLIENT, 0,
1314 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL, NULL);
1315}
1316
1317/*
1318 * We want to handle reconfigure events
1319 */
1320NTSTATUS ctdbd_register_reconfigure(struct ctdbd_connection *conn)
1321{
1322 return register_with_ctdbd(conn, CTDB_SRVID_RECONFIGURE);
1323}
1324
1325/*
1326 call a control on the local node
1327 */
1328NTSTATUS ctdbd_control_local(struct ctdbd_connection *conn, uint32 opcode,
1329 uint64_t srvid, uint32_t flags, TDB_DATA data,
1330 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1331 int *cstatus)
1332{
1333 return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data, mem_ctx, outdata, cstatus);
1334}
1335
1336NTSTATUS ctdb_watch_us(struct ctdbd_connection *conn)
1337{
1338 struct ctdb_client_notify_register reg_data;
1339 size_t struct_len;
1340 NTSTATUS status;
1341 int cstatus;
1342
1343 reg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1344 reg_data.len = 1;
1345 reg_data.notify_data[0] = 0;
1346
1347 struct_len = offsetof(struct ctdb_client_notify_register,
1348 notify_data) + reg_data.len;
1349
1350 status = ctdbd_control_local(
1351 conn, CTDB_CONTROL_REGISTER_NOTIFY, conn->rand_srvid, 0,
1352 make_tdb_data((uint8_t *)&reg_data, struct_len),
1353 NULL, NULL, &cstatus);
1354 if (!NT_STATUS_IS_OK(status)) {
1355 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1356 nt_errstr(status)));
1357 }
1358 return status;
1359}
1360
1361NTSTATUS ctdb_unwatch(struct ctdbd_connection *conn)
1362{
1363 struct ctdb_client_notify_deregister dereg_data;
1364 NTSTATUS status;
1365 int cstatus;
1366
1367 dereg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1368
1369 status = ctdbd_control_local(
1370 conn, CTDB_CONTROL_DEREGISTER_NOTIFY, conn->rand_srvid, 0,
1371 make_tdb_data((uint8_t *)&dereg_data, sizeof(dereg_data)),
1372 NULL, NULL, &cstatus);
1373 if (!NT_STATUS_IS_OK(status)) {
1374 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1375 nt_errstr(status)));
1376 }
1377 return status;
1378}
1379
1380#else
1381
1382NTSTATUS ctdbd_init_connection(TALLOC_CTX *mem_ctx,
1383 struct ctdbd_connection **pconn)
1384{
1385 return NT_STATUS_NOT_IMPLEMENTED;
1386}
1387
1388#endif
Note: See TracBrowser for help on using the repository browser.