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 status;
 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                 uint8_t *reply_data;
1194
1195                 status = ctdb_read_req(conn, 0, talloc_tos(), (void *)&reply);
1196                 if (!NT_STATUS_IS_OK(status)) {
1197                         DEBUG(1, ("ctdb_read_req failed: %s\n",
1198                                   nt_errstr(status)));
1199                         goto fail;
1200                 }
1201
1202                 if (reply->hdr.operation != CTDB_REPLY_CONTROL) {
1203                         DEBUG(1, ("Received invalid reply %u\n",
1204                                   (unsigned)reply->hdr.operation));
1205                         goto fail;
1206                 }
1207
1208                 reqid = reply->hdr.reqid;
1209
1210                 DEBUG(10, ("Received reqid %d\n", (int)reqid));
1211
1212                 for (i=0; i<num_vnns; i++) {
1213                         if (reqid == vnns[i].reqid) {
1214                                 break;
1215                         }
1216                 }
1217                 if (i == num_vnns) {
1218                         DEBUG(1, ("Received unknown reqid number %u\n",
1219                                   (unsigned)reqid));
1220                         goto fail;
1221                 }
1222
1223                 DEBUG(10, ("Found index %u\n", i));
1224
1225                 vnn = &vnns[i];
1226
1227                 DEBUG(10, ("Received vnn %u, vnn->num_srvids %u, datalen %u\n",
1228                            (unsigned)vnn->vnn, vnn->num_srvids,
1229                            (unsigned)reply->datalen));
1230
1231                 if (reply->datalen >= ((vnn->num_srvids+7)/8)) {
1232                         /*
1233                          * Got a real reply
1234                          */
1235                         reply_data = reply->data;
1236                 } else {
1237                         /*
1238                          * Got an error reply
1239                          */
1240                         DEBUG(5, ("Received short reply len %d, status %u, "
1241                                   "errorlen %u\n",
1242                                   (unsigned)reply->datalen,
1243                                   (unsigned)reply->status,
1244                                   (unsigned)reply->errorlen));
1245                         dump_data(5, reply->data, reply->errorlen);
1246
1247                         /*
1248                          * This will trigger everything set to false
1249                          */
1250                         reply_data = NULL;
1251                 }
1252
1253                 for (i=0; i<vnn->num_srvids; i++) {
1254                         int idx = vnn->pid_indexes[i];
1255
1256                         if (pids[i].unique_id ==
1257                             SERVERID_UNIQUE_ID_NOT_TO_VERIFY) {
1258                                 results[idx] = true;
1259                                 continue;
1260                         }
1261                         results[idx] =
1262                                 (reply_data != NULL) &&
1263                                 ((reply_data[i/8] & (1<<(i%8))) != 0);
1264                 }
1265
1266                 TALLOC_FREE(reply);
1267                 num_received += 1;
1268         }
1269
1270         result = true;
1271 fail:
1272         TALLOC_FREE(vnns);
1273         return result;
1274 }
1275
1276 #endif /* HAVE_CTDB_CONTROL_CHECK_SRVIDS_DECL */
1277
1278 /*
1279  * Get a db path
1280  */
1281 char *ctdbd_dbpath(struct ctdbd_connection *conn,
1282                    TALLOC_CTX *mem_ctx, uint32_t db_id)
1283 {
1284         NTSTATUS status;
1285         TDB_DATA data;
1286         int32_t cstatus;
1287
1288         data.dptr = (uint8_t*)&db_id;
1289         data.dsize = sizeof(db_id);
1290
1291         status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1292                                CTDB_CONTROL_GETDBPATH, 0, 0, data,
1293                                mem_ctx, &data, &cstatus);
1294         if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
1295                 DEBUG(0,(__location__ " ctdb_control for getdbpath failed\n"));
1296                 return NULL;
1297         }
1298
1299         return (char *)data.dptr;
1300 }
1301
1302 /*
1303  * attach to a ctdb database
1304  */
1305 NTSTATUS ctdbd_db_attach(struct ctdbd_connection *conn,
1306                          const char *name, uint32_t *db_id, int tdb_flags)
1307 {
1308         NTSTATUS status;
1309         TDB_DATA data;
1310         int32_t cstatus;
1311         bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
1312
1313         data = string_term_tdb_data(name);
1314
1315         status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1316                                persistent
1317                                ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
1318                                : CTDB_CONTROL_DB_ATTACH,
1319                                tdb_flags, 0, data, NULL, &data, &cstatus);
1320         if (!NT_STATUS_IS_OK(status)) {
1321                 DEBUG(0, (__location__ " ctdb_control for db_attach "
1322                           "failed: %s\n", nt_errstr(status)));
1323                 return status;
1324         }
1325
1326         if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
1327                 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
1328                 return NT_STATUS_INTERNAL_ERROR;
1329         }
1330
1331         *db_id = *(uint32_t *)data.dptr;
1332         talloc_free(data.dptr);
1333
1334         if (!(tdb_flags & TDB_SEQNUM)) {
1335                 return NT_STATUS_OK;
1336         }
1337
1338         data.dptr = (uint8_t *)db_id;
1339         data.dsize = sizeof(*db_id);
1340
1341         status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1342                                CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
1343                                NULL, NULL, &cstatus);
1344         if (!NT_STATUS_IS_OK(status) || cstatus != 0) {
1345                 DEBUG(0,(__location__ " ctdb_control for enable seqnum "
1346                          "failed\n"));
1347                 return NT_STATUS_IS_OK(status) ? NT_STATUS_INTERNAL_ERROR :
1348                         status;
1349         }
1350
1351         return NT_STATUS_OK;
1352 }
1353
1354 /*
1355  * force the migration of a record to this node
1356  */
1357 NTSTATUS ctdbd_migrate(struct ctdbd_connection *conn, uint32_t db_id,
1358                        TDB_DATA key)
1359 {
1360         struct ctdb_req_call req;
1361         struct ctdb_reply_call *reply;
1362         NTSTATUS status;
1363
1364         ZERO_STRUCT(req);
1365
1366         req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
1367         req.hdr.ctdb_magic   = CTDB_MAGIC;
1368         req.hdr.ctdb_version = CTDB_VERSION;
1369         req.hdr.operation    = CTDB_REQ_CALL;
1370         req.hdr.reqid        = ctdbd_next_reqid(conn);
1371         req.flags            = CTDB_IMMEDIATE_MIGRATION;
1372         req.callid           = CTDB_NULL_FUNC;
1373         req.db_id            = db_id;
1374         req.keylen           = key.dsize;
1375
1376         DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
1377         ctdb_packet_dump(&req.hdr);
1378
1379         status = ctdb_packet_send(
1380                 conn->pkt, 2,
1381                 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
1382                 data_blob_const(key.dptr, key.dsize));
1383
1384         if (!NT_STATUS_IS_OK(status)) {
1385                 DEBUG(3, ("ctdb_packet_send failed: %s\n", nt_errstr(status)));
1386                 return status;
1387         }
1388
1389         status = ctdb_packet_flush(conn->pkt);
1390
1391         if (!NT_STATUS_IS_OK(status)) {
1392                 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
1393                 cluster_fatal("cluster dispatch daemon control write error\n");
1394         }
1395
1396         status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
1397
1398         if (!NT_STATUS_IS_OK(status)) {
1399                 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
1400                 goto fail;
1401         }
1402
1403         if (reply->hdr.operation != CTDB_REPLY_CALL) {
1404                 DEBUG(0, ("received invalid reply\n"));
1405                 status = NT_STATUS_INTERNAL_ERROR;
1406                 goto fail;
1407         }
1408
1409         status = NT_STATUS_OK;
1410  fail:
1411
1412         TALLOC_FREE(reply);
1413         return status;
1414 }
1415
1416 /*
1417  * remotely fetch a record (read-only)
1418  */
1419 NTSTATUS ctdbd_fetch(struct ctdbd_connection *conn, uint32_t db_id,
1420                      TDB_DATA key, TALLOC_CTX *mem_ctx, TDB_DATA *data,
1421                      bool local_copy)
1422 {
1423         struct ctdb_req_call req;
1424         struct ctdb_reply_call *reply;
1425         NTSTATUS status;
1426         uint32_t flags;
1427
1428 #ifdef HAVE_CTDB_WANT_READONLY_DECL
1429         flags = local_copy ? CTDB_WANT_READONLY : 0;
1430 #else
1431         flags = 0;
1432 #endif
1433
1434         ZERO_STRUCT(req);
1435
1436         req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
1437         req.hdr.ctdb_magic   = CTDB_MAGIC;
1438         req.hdr.ctdb_version = CTDB_VERSION;
1439         req.hdr.operation    = CTDB_REQ_CALL;
1440         req.hdr.reqid        = ctdbd_next_reqid(conn);
1441         req.flags            = flags;
1442         req.callid           = CTDB_FETCH_FUNC;
1443         req.db_id            = db_id;
1444         req.keylen           = key.dsize;
1445
1446         status = ctdb_packet_send(
1447                 conn->pkt, 2,
1448                 data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
1449                 data_blob_const(key.dptr, key.dsize));
1450
1451         if (!NT_STATUS_IS_OK(status)) {
1452                 DEBUG(3, ("ctdb_packet_send failed: %s\n", nt_errstr(status)));
1453                 return status;
1454         }
1455
1456         status = ctdb_packet_flush(conn->pkt);
1457
1458         if (!NT_STATUS_IS_OK(status)) {
1459                 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
1460                 cluster_fatal("cluster dispatch daemon control write error\n");
1461         }
1462
1463         status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
1464
1465         if (!NT_STATUS_IS_OK(status)) {
1466                 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
1467                 goto fail;
1468         }
1469
1470         if (reply->hdr.operation != CTDB_REPLY_CALL) {
1471                 DEBUG(0, ("received invalid reply\n"));
1472                 status = NT_STATUS_INTERNAL_ERROR;
1473                 goto fail;
1474         }
1475
1476         data->dsize = reply->datalen;
1477         if (data->dsize == 0) {
1478                 data->dptr = NULL;
1479                 goto done;
1480         }
1481
1482         data->dptr = (uint8 *)talloc_memdup(mem_ctx, &reply->data[0],
1483                                             reply->datalen);
1484         if (data->dptr == NULL) {
1485                 DEBUG(0, ("talloc failed\n"));
1486                 status = NT_STATUS_NO_MEMORY;
1487                 goto fail;
1488         }
1489
1490  done:
1491         status = NT_STATUS_OK;
1492  fail:
1493         TALLOC_FREE(reply);
1494         return status;
1495 }
1496
1497 struct ctdbd_traverse_state {
1498         void (*fn)(TDB_DATA key, TDB_DATA data, void *private_data);
1499         void *private_data;
1500 };
1501
1502 /*
1503  * Handle a traverse record coming in on the ctdbd connection
1504  */
1505
1506 static NTSTATUS ctdb_traverse_handler(uint8_t *buf, size_t length,
1507                                       void *private_data)
1508 {
1509         struct ctdbd_traverse_state *state =
1510                 (struct ctdbd_traverse_state *)private_data;
1511
1512         struct ctdb_req_message *m;
1513         struct ctdb_rec_data *d;
1514         TDB_DATA key, data;
1515
1516         m = (struct ctdb_req_message *)buf;
1517
1518         if (length < sizeof(*m) || m->hdr.length != length) {
1519                 DEBUG(0, ("Got invalid message of length %d\n", (int)length));
1520                 TALLOC_FREE(buf);
1521                 return NT_STATUS_UNEXPECTED_IO_ERROR;
1522         }
1523
1524         d = (struct ctdb_rec_data *)&m->data[0];
1525         if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1526                 DEBUG(0, ("Got invalid traverse data of length %d\n",
1527                           (int)m->datalen));
1528                 TALLOC_FREE(buf);
1529                 return NT_STATUS_UNEXPECTED_IO_ERROR;
1530         }
1531
1532         key.dsize = d->keylen;
1533         key.dptr  = &d->data[0];
1534         data.dsize = d->datalen;
1535         data.dptr = &d->data[d->keylen];
1536
1537         if (key.dsize == 0 && data.dsize == 0) {
1538                 /* end of traverse */
1539                 return NT_STATUS_END_OF_FILE;
1540         }
1541
1542         if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1543                 DEBUG(0, ("Got invalid ltdb header length %d\n",
1544                           (int)data.dsize));
1545                 TALLOC_FREE(buf);
1546                 return NT_STATUS_UNEXPECTED_IO_ERROR;
1547         }
1548         data.dsize -= sizeof(struct ctdb_ltdb_header);
1549         data.dptr += sizeof(struct ctdb_ltdb_header);
1550
1551         if (state->fn) {
1552                 state->fn(key, data, state->private_data);
1553         }
1554
1555         TALLOC_FREE(buf);
1556         return NT_STATUS_OK;
1557 }
1558
1559 /*
1560   Traverse a ctdb database. This uses a kind-of hackish way to open a second
1561   connection to ctdbd to avoid the hairy recursive and async problems with
1562   everything in-line.
1563 */
1564
1565 NTSTATUS ctdbd_traverse(uint32_t db_id,
1566                         void (*fn)(TDB_DATA key, TDB_DATA data,
1567                                    void *private_data),
1568                         void *private_data)
1569 {
1570         struct ctdbd_connection *conn;
1571         NTSTATUS status;
1572
1573         TDB_DATA data;
1574         struct ctdb_traverse_start t;
1575         int cstatus;
1576         struct ctdbd_traverse_state state;
1577
1578         become_root();
1579         status = ctdbd_init_connection(NULL, &conn);
1580         unbecome_root();
1581         if (!NT_STATUS_IS_OK(status)) {
1582                 DEBUG(0, ("ctdbd_init_connection failed: %s\n",
1583                           nt_errstr(status)));
1584                 return status;
1585         }
1586
1587         t.db_id = db_id;
1588         t.srvid = conn->rand_srvid;
1589         t.reqid = ctdbd_next_reqid(conn);
1590
1591         data.dptr = (uint8_t *)&t;
1592         data.dsize = sizeof(t);
1593
1594         status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1595                                CTDB_CONTROL_TRAVERSE_START, conn->rand_srvid, 0,
1596                                data, NULL, NULL, &cstatus);
1597
1598         if (!NT_STATUS_IS_OK(status) || (cstatus != 0)) {
1599
1600                 DEBUG(0,("ctdbd_control failed: %s, %d\n", nt_errstr(status),
1601                          cstatus));
1602
1603                 if (NT_STATUS_IS_OK(status)) {
1604                         /*
1605                          * We need a mapping here
1606                          */
1607                         status = NT_STATUS_UNSUCCESSFUL;
1608                 }
1609                 goto done;
1610         }
1611
1612         state.fn = fn;
1613         state.private_data = private_data;
1614
1615         while (True) {
1616
1617                 status = NT_STATUS_OK;
1618
1619                 if (ctdb_packet_handler(conn->pkt, ctdb_req_complete,
1620                                    ctdb_traverse_handler, &state, &status)) {
1621
1622                         if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1623                                 status = NT_STATUS_OK;
1624                                 break;
1625                         }
1626
1627                         /*
1628                          * There might be more in the queue
1629                          */
1630                         continue;
1631                 }
1632
1633                 if (!NT_STATUS_IS_OK(status)) {
1634                         break;
1635                 }
1636
1637                 status = ctdb_packet_fd_read_sync(conn->pkt);
1638
1639                 if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
1640                         /*
1641                          * There might be more in the queue
1642                          */
1643                         continue;
1644                 }
1645
1646                 if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1647                         status = NT_STATUS_OK;
1648                         break;
1649                 }
1650
1651                 if (!NT_STATUS_IS_OK(status)) {
1652                         DEBUG(0, ("ctdb_packet_fd_read_sync failed: %s\n", nt_errstr(status)));
1653                         cluster_fatal("ctdbd died\n");
1654                 }
1655         }
1656
1657  done:
1658         TALLOC_FREE(conn);
1659         return status;
1660 }
1661
1662 /*
1663    This is used to canonicalize a ctdb_sock_addr structure.
1664 */
1665 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1666                                       struct sockaddr_storage *out)
1667 {
1668         memcpy(out, in, sizeof (*out));
1669
1670 #ifdef HAVE_IPV6
1671         if (in->ss_family == AF_INET6) {
1672                 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1673                 const struct sockaddr_in6 *in6 =
1674                         (const struct sockaddr_in6 *)in;
1675                 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1676                 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1677                         memset(out, 0, sizeof(*out));
1678 #ifdef HAVE_SOCK_SIN_LEN
1679                         out4->sin_len = sizeof(*out);
1680 #endif
1681                         out4->sin_family = AF_INET;
1682                         out4->sin_port   = in6->sin6_port;
1683                         memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr[12], 4);
1684                 }
1685         }
1686 #endif
1687 }
1688
1689 /*
1690  * Register us as a server for a particular tcp connection
1691  */
1692
1693 NTSTATUS ctdbd_register_ips(struct ctdbd_connection *conn,
1694                             const struct sockaddr_storage *_server,
1695                             const struct sockaddr_storage *_client,
1696                             void (*release_ip_handler)(const char *ip_addr,
1697                                                        void *private_data),
1698                             void *private_data)
1699 {
1700         /*
1701          * we still use ctdb_control_tcp for ipv4
1702          * because we want to work against older ctdb
1703          * versions at runtime
1704          */
1705         struct ctdb_control_tcp p4;
1706 #ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1707         struct ctdb_control_tcp_addr p;
1708 #endif
1709         TDB_DATA data;
1710         NTSTATUS status;
1711         struct sockaddr_storage client;
1712         struct sockaddr_storage server;
1713
1714         /*
1715          * Only one connection so far
1716          */
1717         SMB_ASSERT(conn->release_ip_handler == NULL);
1718
1719         smbd_ctdb_canonicalize_ip(_client, &client);
1720         smbd_ctdb_canonicalize_ip(_server, &server);
1721
1722         switch (client.ss_family) {
1723         case AF_INET:
1724                 memcpy(&p4.dest, &server, sizeof(p4.dest));
1725                 memcpy(&p4.src, &client, sizeof(p4.src));
1726                 data.dptr = (uint8_t *)&p4;
1727                 data.dsize = sizeof(p4);
1728                 break;
1729 #ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1730         case AF_INET6:
1731                 memcpy(&p.dest.ip6, &server, sizeof(p.dest.ip6));
1732                 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1733                 data.dptr = (uint8_t *)&p;
1734                 data.dsize = sizeof(p);
1735                 break;
1736 #endif
1737         default:
1738                 return NT_STATUS_INTERNAL_ERROR;
1739         }
1740
1741         conn->release_ip_handler = release_ip_handler;
1742         conn->release_ip_priv = private_data;
1743
1744         /*
1745          * We want to be told about IP releases
1746          */
1747
1748         status = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP);
1749         if (!NT_STATUS_IS_OK(status)) {
1750                 return status;
1751         }
1752
1753         /*
1754          * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1755          * can send an extra ack to trigger a reset for our client, so it
1756          * immediately reconnects
1757          */
1758         return ctdbd_control(conn, CTDB_CURRENT_NODE,
1759                              CTDB_CONTROL_TCP_CLIENT, 0,
1760                              CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL, NULL);
1761 }
1762
1763 /*
1764  * We want to handle reconfigure events
1765  */
1766 NTSTATUS ctdbd_register_reconfigure(struct ctdbd_connection *conn)
1767 {
1768         return register_with_ctdbd(conn, CTDB_SRVID_RECONFIGURE);
1769 }
1770
1771 /*
1772   call a control on the local node
1773  */
1774 NTSTATUS ctdbd_control_local(struct ctdbd_connection *conn, uint32_t opcode,
1775                              uint64_t srvid, uint32_t flags, TDB_DATA data,
1776                              TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1777                              int *cstatus)
1778 {
1779         return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data, mem_ctx, outdata, cstatus);
1780 }
1781
1782 NTSTATUS ctdb_watch_us(struct ctdbd_connection *conn)
1783 {
1784         struct ctdb_client_notify_register reg_data;
1785         size_t struct_len;
1786         NTSTATUS status;
1787         int cstatus;
1788
1789         reg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1790         reg_data.len = 1;
1791         reg_data.notify_data[0] = 0;
1792
1793         struct_len = offsetof(struct ctdb_client_notify_register,
1794                               notify_data) + reg_data.len;
1795
1796         status = ctdbd_control_local(
1797                 conn, CTDB_CONTROL_REGISTER_NOTIFY, conn->rand_srvid, 0,
1798                 make_tdb_data((uint8_t *)&reg_data, struct_len),
1799                 NULL, NULL, &cstatus);
1800         if (!NT_STATUS_IS_OK(status)) {
1801                 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1802                           nt_errstr(status)));
1803         }
1804         return status;
1805 }
1806
1807 NTSTATUS ctdb_unwatch(struct ctdbd_connection *conn)
1808 {
1809         struct ctdb_client_notify_deregister dereg_data;
1810         NTSTATUS status;
1811         int cstatus;
1812
1813         dereg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1814
1815         status = ctdbd_control_local(
1816                 conn, CTDB_CONTROL_DEREGISTER_NOTIFY, conn->rand_srvid, 0,
1817                 make_tdb_data((uint8_t *)&dereg_data, sizeof(dereg_data)),
1818                 NULL, NULL, &cstatus);
1819         if (!NT_STATUS_IS_OK(status)) {
1820                 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1821                           nt_errstr(status)));
1822         }
1823         return status;
1824 }
1825
1826 #else
1827
1828 NTSTATUS ctdbd_messaging_send_blob(struct ctdbd_connection *conn,
1829                                    uint32_t dst_vnn, uint64_t dst_srvid,
1830                                    const uint8_t *buf, size_t buflen)
1831 {
1832         return NT_STATUS_NOT_IMPLEMENTED;
1833 }
1834
1835 #endif