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