source3/lib/ctdbd_conn.c

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