source3/lib/ctdbd_conn.c

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