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