source3/lib/ctdbd_conn.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    Samba internal messaging functions
   4    Copyright (C) 2007 by Volker Lendecke
   5    Copyright (C) 2007 by Andrew Tridgell
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 */
  20
  21 #include "includes.h"
  22 #include "util_tdb.h"
  23 #include "serverid.h"
  24 #include "ctdbd_conn.h"
  25 #include "system/select.h"
  26 #include "lib/util/sys_rw_data.h"
  27 #include "lib/util/iov_buf.h"
  28
  29 #include "messages.h"
  30
  31 /* paths to these include files come from --with-ctdb= in configure */
  32
  33 #include "ctdb_private.h"
  34
  35 struct ctdbd_srvid_cb {
  36         uint64_t srvid;
  37         int (*cb)(uint32_t src_vnn, uint32_t dst_vnn,
  38                   uint64_t dst_srvid,
  39                   const uint8_t *msg, size_t msglen,
  40                   void *private_data);
  41         void *private_data;
  42 };
  43
  44 struct ctdbd_connection {
  45         const char *sockname;   /* Needed in ctdbd_traverse */
  46         struct messaging_context *msg_ctx;
  47         uint32_t reqid;
  48         uint32_t our_vnn;
  49         uint64_t rand_srvid;
  50         struct ctdbd_srvid_cb *callbacks;
  51         int fd;
  52         struct tevent_fd *fde;
  53         int timeout;
  54 };
  55
  56 static uint32_t ctdbd_next_reqid(struct ctdbd_connection *conn)
  57 {
  58         conn->reqid += 1;
  59         if (conn->reqid == 0) {
  60                 conn->reqid += 1;
  61         }
  62         return conn->reqid;
  63 }
  64
  65 static int ctdbd_control(struct ctdbd_connection *conn,
  66                          uint32_t vnn, uint32_t opcode,
  67                          uint64_t srvid, uint32_t flags,
  68                          TDB_DATA data,
  69                          TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
  70                          int *cstatus);
  71
  72 /*
  73  * exit on fatal communications errors with the ctdbd daemon
  74  */
  75 static void cluster_fatal(const char *why)
  76 {
  77         DEBUG(0,("cluster fatal event: %s - exiting immediately\n", why));
  78         /* we don't use smb_panic() as we don't want to delay to write
  79            a core file. We need to release this process id immediately
  80            so that someone else can take over without getting sharing
  81            violations */
  82         _exit(1);
  83 }
  84
  85 /*
  86  *
  87  */
  88 static void ctdb_packet_dump(struct ctdb_req_header *hdr)
  89 {
  90         if (DEBUGLEVEL < 11) {
  91                 return;
  92         }
  93         DEBUGADD(11, ("len=%d, magic=%x, vers=%d, gen=%d, op=%d, reqid=%d\n",
  94                       (int)hdr->length, (int)hdr->ctdb_magic,
  95                       (int)hdr->ctdb_version, (int)hdr->generation,
  96                       (int)hdr->operation, (int)hdr->reqid));
  97 }
  98
  99 /*
 100  * Register a srvid with ctdbd
 101  */
 102 int register_with_ctdbd(struct ctdbd_connection *conn, uint64_t srvid,
 103                         int (*cb)(uint32_t src_vnn, uint32_t dst_vnn,
 104                                   uint64_t dst_srvid,
 105                                   const uint8_t *msg, size_t msglen,
 106                                   void *private_data),
 107                         void *private_data)
 108 {
 109
 110         int ret, cstatus;
 111         size_t num_callbacks;
 112         struct ctdbd_srvid_cb *tmp;
 113
 114         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
 115                             CTDB_CONTROL_REGISTER_SRVID, srvid, 0,
 116                             tdb_null, NULL, NULL, &cstatus);
 117         if (ret != 0) {
 118                 return ret;
 119         }
 120
 121         num_callbacks = talloc_array_length(conn->callbacks);
 122
 123         tmp = talloc_realloc(conn, conn->callbacks, struct ctdbd_srvid_cb,
 124                              num_callbacks + 1);
 125         if (tmp == NULL) {
 126                 return ENOMEM;
 127         }
 128         conn->callbacks = tmp;
 129
 130         conn->callbacks[num_callbacks] = (struct ctdbd_srvid_cb) {
 131                 .srvid = srvid, .cb = cb, .private_data = private_data
 132         };
 133
 134         return 0;
 135 }
 136
 137 static int ctdbd_msg_call_back(struct ctdbd_connection *conn,
 138                                struct ctdb_req_message_old *msg)
 139 {
 140         size_t msg_len;
 141         size_t i, num_callbacks;
 142
 143         msg_len = msg->hdr.length;
 144         if (msg_len < offsetof(struct ctdb_req_message_old, data)) {
 145                 DEBUG(10, ("%s: len %u too small\n", __func__,
 146                            (unsigned)msg_len));
 147                 return 0;
 148         }
 149         msg_len -= offsetof(struct ctdb_req_message_old, data);
 150
 151         if (msg_len < msg->datalen) {
 152                 DEBUG(10, ("%s: msg_len=%u < msg->datalen=%u\n", __func__,
 153                            (unsigned)msg_len, (unsigned)msg->datalen));
 154                 return 0;
 155         }
 156
 157         num_callbacks = talloc_array_length(conn->callbacks);
 158
 159         for (i=0; i<num_callbacks; i++) {
 160                 struct ctdbd_srvid_cb *cb = &conn->callbacks[i];
 161
 162                 if ((cb->srvid == msg->srvid) && (cb->cb != NULL)) {
 163                         int ret;
 164
 165                         ret = cb->cb(msg->hdr.srcnode, msg->hdr.destnode,
 166                                      msg->srvid, msg->data, msg->datalen,
 167                                      cb->private_data);
 168                         if (ret != 0) {
 169                                 return ret;
 170                         }
 171                 }
 172         }
 173         return 0;
 174 }
 175
 176 /*
 177  * get our vnn from the cluster
 178  */
 179 static int get_cluster_vnn(struct ctdbd_connection *conn, uint32_t *vnn)
 180 {
 181         int32_t cstatus=-1;
 182         int ret;
 183         ret = ctdbd_control(conn,
 184                             CTDB_CURRENT_NODE, CTDB_CONTROL_GET_PNN, 0, 0,
 185                             tdb_null, NULL, NULL, &cstatus);
 186         if (ret != 0) {
 187                 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
 188                 return ret;
 189         }
 190         *vnn = (uint32_t)cstatus;
 191         return ret;
 192 }
 193
 194 /*
 195  * Are we active (i.e. not banned or stopped?)
 196  */
 197 static bool ctdbd_working(struct ctdbd_connection *conn, uint32_t vnn)
 198 {
 199         int32_t cstatus=-1;
 200         TDB_DATA outdata;
 201         struct ctdb_node_map_old *m;
 202         uint32_t failure_flags;
 203         bool ok = false;
 204         int i, ret;
 205
 206         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
 207                             CTDB_CONTROL_GET_NODEMAP, 0, 0,
 208                             tdb_null, talloc_tos(), &outdata, &cstatus);
 209         if (ret != 0) {
 210                 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
 211                 return false;
 212         }
 213         if ((cstatus != 0) || (outdata.dptr == NULL)) {
 214                 DEBUG(2, ("Received invalid ctdb data\n"));
 215                 return false;
 216         }
 217
 218         m = (struct ctdb_node_map_old *)outdata.dptr;
 219
 220         for (i=0; i<m->num; i++) {
 221                 if (vnn == m->nodes[i].pnn) {
 222                         break;
 223                 }
 224         }
 225
 226         if (i == m->num) {
 227                 DEBUG(2, ("Did not find ourselves (node %d) in nodemap\n",
 228                           (int)vnn));
 229                 goto fail;
 230         }
 231
 232         failure_flags = NODE_FLAGS_BANNED | NODE_FLAGS_DISCONNECTED
 233                 | NODE_FLAGS_PERMANENTLY_DISABLED | NODE_FLAGS_STOPPED;
 234
 235         if ((m->nodes[i].flags & failure_flags) != 0) {
 236                 DEBUG(2, ("Node has status %x, not active\n",
 237                           (int)m->nodes[i].flags));
 238                 goto fail;
 239         }
 240
 241         ok = true;
 242 fail:
 243         TALLOC_FREE(outdata.dptr);
 244         return ok;
 245 }
 246
 247 uint32_t ctdbd_vnn(const struct ctdbd_connection *conn)
 248 {
 249         return conn->our_vnn;
 250 }
 251
 252 /*
 253  * Get us a ctdb connection
 254  */
 255
 256 static int ctdbd_connect(const char *sockname, int *pfd)
 257 {
 258         struct sockaddr_un addr = { 0, };
 259         int fd;
 260         socklen_t salen;
 261         size_t namelen;
 262
 263         fd = socket(AF_UNIX, SOCK_STREAM, 0);
 264         if (fd == -1) {
 265                 int err = errno;
 266                 DEBUG(3, ("Could not create socket: %s\n", strerror(err)));
 267                 return err;
 268         }
 269
 270         addr.sun_family = AF_UNIX;
 271
 272         namelen = strlcpy(addr.sun_path, sockname, sizeof(addr.sun_path));
 273         if (namelen >= sizeof(addr.sun_path)) {
 274                 DEBUG(3, ("%s: Socket name too long: %s\n", __func__,
 275                           sockname));
 276                 close(fd);
 277                 return ENAMETOOLONG;
 278         }
 279
 280         salen = sizeof(struct sockaddr_un);
 281
 282         if (connect(fd, (struct sockaddr *)(void *)&addr, salen) == -1) {
 283                 int err = errno;
 284                 DEBUG(1, ("connect(%s) failed: %s\n", sockname,
 285                           strerror(err)));
 286                 close(fd);
 287                 return err;
 288         }
 289
 290         *pfd = fd;
 291         return 0;
 292 }
 293
 294 static int ctdb_read_packet(int fd, int timeout, TALLOC_CTX *mem_ctx,
 295                             struct ctdb_req_header **result)
 296 {
 297         struct ctdb_req_header *req;
 298         int ret, revents;
 299         uint32_t msglen;
 300         ssize_t nread;
 301
 302         if (timeout != -1) {
 303                 ret = poll_intr_one_fd(fd, POLLIN, timeout, &revents);
 304                 if (ret == -1) {
 305                         return errno;
 306                 }
 307                 if (ret == 0) {
 308                         return ETIMEDOUT;
 309                 }
 310                 if (ret != 1) {
 311                         return EIO;
 312                 }
 313         }
 314
 315         nread = read_data(fd, &msglen, sizeof(msglen));
 316         if (nread == -1) {
 317                 return errno;
 318         }
 319         if (nread == 0) {
 320                 return EIO;
 321         }
 322
 323         if (msglen < sizeof(struct ctdb_req_header)) {
 324                 return EIO;
 325         }
 326
 327         req = talloc_size(mem_ctx, msglen);
 328         if (req == NULL) {
 329                 return ENOMEM;
 330         }
 331         talloc_set_name_const(req, "struct ctdb_req_header");
 332
 333         req->length = msglen;
 334
 335         nread = read_data(fd, ((char *)req) + sizeof(msglen),
 336                           msglen - sizeof(msglen));
 337         if (nread == -1) {
 338                 TALLOC_FREE(req);
 339                 return errno;
 340         }
 341         if (nread == 0) {
 342                 TALLOC_FREE(req);
 343                 return EIO;
 344         }
 345
 346         *result = req;
 347         return 0;
 348 }
 349
 350 /*
 351  * Read a full ctdbd request. If we have a messaging context, defer incoming
 352  * messages that might come in between.
 353  */
 354
 355 static int ctdb_read_req(struct ctdbd_connection *conn, uint32_t reqid,
 356                          TALLOC_CTX *mem_ctx, struct ctdb_req_header **result)
 357 {
 358         struct ctdb_req_header *hdr;
 359         int ret;
 360
 361  next_pkt:
 362
 363         ret = ctdb_read_packet(conn->fd, conn->timeout, mem_ctx, &hdr);
 364         if (ret != 0) {
 365                 DEBUG(0, ("ctdb_read_packet failed: %s\n", strerror(ret)));
 366                 cluster_fatal("ctdbd died\n");
 367         }
 368
 369         DEBUG(11, ("Received ctdb packet\n"));
 370         ctdb_packet_dump(hdr);
 371
 372         if (hdr->operation == CTDB_REQ_MESSAGE) {
 373                 struct ctdb_req_message_old *msg = (struct ctdb_req_message_old *)hdr;
 374
 375                 if (conn->msg_ctx == NULL) {
 376                         DEBUG(1, ("Got a message without having a msg ctx, "
 377                                   "dropping msg %llu\n",
 378                                   (long long unsigned)msg->srvid));
 379                         TALLOC_FREE(hdr);
 380                         goto next_pkt;
 381                 }
 382
 383                 ret = ctdbd_msg_call_back(conn, msg);
 384                 if (ret != 0) {
 385                         TALLOC_FREE(hdr);
 386                         return ret;
 387                 }
 388
 389                 TALLOC_FREE(hdr);
 390                 goto next_pkt;
 391         }
 392
 393         if ((reqid != 0) && (hdr->reqid != reqid)) {
 394                 /* we got the wrong reply */
 395                 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
 396                          "been %u\n", hdr->reqid, reqid));
 397                 TALLOC_FREE(hdr);
 398                 goto next_pkt;
 399         }
 400
 401         *result = talloc_move(mem_ctx, &hdr);
 402
 403         return 0;
 404 }
 405
 406 static int ctdbd_connection_destructor(struct ctdbd_connection *c)
 407 {
 408         TALLOC_FREE(c->fde);
 409         if (c->fd != -1) {
 410                 close(c->fd);
 411                 c->fd = -1;
 412         }
 413         return 0;
 414 }
 415 /*
 416  * Get us a ctdbd connection
 417  */
 418
 419 static int ctdbd_init_connection(TALLOC_CTX *mem_ctx,
 420                                  const char *sockname, int timeout,
 421                                  struct ctdbd_connection **pconn)
 422 {
 423         struct ctdbd_connection *conn;
 424         int ret;
 425
 426         if (!(conn = talloc_zero(mem_ctx, struct ctdbd_connection))) {
 427                 DEBUG(0, ("talloc failed\n"));
 428                 return ENOMEM;
 429         }
 430
 431         conn->sockname = talloc_strdup(conn, sockname);
 432         if (conn->sockname == NULL) {
 433                 DBG_ERR("talloc failed\n");
 434                 ret = ENOMEM;
 435                 goto fail;
 436         }
 437
 438         conn->timeout = timeout;
 439
 440         if (conn->timeout == 0) {
 441                 conn->timeout = -1;
 442         }
 443
 444         ret = ctdbd_connect(conn->sockname, &conn->fd);
 445         if (ret != 0) {
 446                 DEBUG(1, ("ctdbd_connect failed: %s\n", strerror(ret)));
 447                 goto fail;
 448         }
 449         talloc_set_destructor(conn, ctdbd_connection_destructor);
 450
 451         ret = get_cluster_vnn(conn, &conn->our_vnn);
 452
 453         if (ret != 0) {
 454                 DEBUG(10, ("get_cluster_vnn failed: %s\n", strerror(ret)));
 455                 goto fail;
 456         }
 457
 458         if (!ctdbd_working(conn, conn->our_vnn)) {
 459                 DEBUG(2, ("Node is not working, can not connect\n"));
 460                 ret = EIO;
 461                 goto fail;
 462         }
 463
 464         generate_random_buffer((unsigned char *)&conn->rand_srvid,
 465                                sizeof(conn->rand_srvid));
 466
 467         ret = register_with_ctdbd(conn, conn->rand_srvid, NULL, NULL);
 468
 469         if (ret != 0) {
 470                 DEBUG(5, ("Could not register random srvid: %s\n",
 471                           strerror(ret)));
 472                 goto fail;
 473         }
 474
 475         *pconn = conn;
 476         return 0;
 477
 478  fail:
 479         TALLOC_FREE(conn);
 480         return ret;
 481 }
 482
 483 /*
 484  * Get us a ctdbd connection and register us as a process
 485  */
 486
 487 int ctdbd_messaging_connection(TALLOC_CTX *mem_ctx,
 488                                const char *sockname, int timeout,
 489                                struct ctdbd_connection **pconn)
 490 {
 491         struct ctdbd_connection *conn;
 492         int ret;
 493
 494         ret = ctdbd_init_connection(mem_ctx, sockname, timeout, &conn);
 495
 496         if (ret != 0) {
 497                 return ret;
 498         }
 499
 500         ret = register_with_ctdbd(conn, MSG_SRVID_SAMBA, NULL, NULL);
 501         if (ret != 0) {
 502                 goto fail;
 503         }
 504
 505         *pconn = conn;
 506         return 0;
 507
 508  fail:
 509         TALLOC_FREE(conn);
 510         return ret;
 511 }
 512
 513 struct messaging_context *ctdb_conn_msg_ctx(struct ctdbd_connection *conn)
 514 {
 515         return conn->msg_ctx;
 516 }
 517
 518 int ctdbd_conn_get_fd(struct ctdbd_connection *conn)
 519 {
 520         return conn->fd;
 521 }
 522
 523 /*
 524  * Packet handler to receive and handle a ctdb message
 525  */
 526 static int ctdb_handle_message(struct ctdbd_connection *conn,
 527                                struct ctdb_req_header *hdr)
 528 {
 529         struct ctdb_req_message_old *msg;
 530
 531         if (hdr->operation != CTDB_REQ_MESSAGE) {
 532                 DEBUG(0, ("Received async msg of type %u, discarding\n",
 533                           hdr->operation));
 534                 return EINVAL;
 535         }
 536
 537         msg = (struct ctdb_req_message_old *)hdr;
 538
 539         ctdbd_msg_call_back(conn, msg);
 540
 541         return 0;
 542 }
 543
 544 /*
 545  * The ctdbd socket is readable asynchronuously
 546  */
 547
 548 static void ctdbd_socket_handler(struct tevent_context *event_ctx,
 549                                  struct tevent_fd *event,
 550                                  uint16_t flags,
 551                                  void *private_data)
 552 {
 553         struct ctdbd_connection *conn = talloc_get_type_abort(
 554                 private_data, struct ctdbd_connection);
 555         struct ctdb_req_header *hdr = NULL;
 556         int ret;
 557
 558         ret = ctdb_read_packet(conn->fd, conn->timeout, talloc_tos(), &hdr);
 559         if (ret != 0) {
 560                 DEBUG(0, ("ctdb_read_packet failed: %s\n", strerror(ret)));
 561                 cluster_fatal("ctdbd died\n");
 562         }
 563
 564         ret = ctdb_handle_message(conn, hdr);
 565
 566         TALLOC_FREE(hdr);
 567
 568         if (ret != 0) {
 569                 DEBUG(10, ("could not handle incoming message: %s\n",
 570                            strerror(ret)));
 571         }
 572 }
 573
 574 /*
 575  * Prepare a ctdbd connection to receive messages
 576  */
 577
 578 int ctdbd_register_msg_ctx(struct ctdbd_connection *conn,
 579                            struct messaging_context *msg_ctx,
 580                            struct tevent_context *ev)
 581 {
 582         SMB_ASSERT(conn->msg_ctx == NULL);
 583         SMB_ASSERT(conn->fde == NULL);
 584
 585         conn->fde = tevent_add_fd(ev, conn, conn->fd, TEVENT_FD_READ,
 586                                   ctdbd_socket_handler, conn);
 587         if (conn->fde == NULL) {
 588                 DEBUG(0, ("event_add_fd failed\n"));
 589                 return ENOMEM;
 590         }
 591
 592         conn->msg_ctx = msg_ctx;
 593
 594         return 0;
 595 }
 596
 597 int ctdbd_messaging_send_iov(struct ctdbd_connection *conn,
 598                              uint32_t dst_vnn, uint64_t dst_srvid,
 599                              const struct iovec *iov, int iovlen)
 600 {
 601         struct ctdb_req_message_old r;
 602         struct iovec iov2[iovlen+1];
 603         size_t buflen = iov_buflen(iov, iovlen);
 604         ssize_t nwritten;
 605
 606         r.hdr.length = offsetof(struct ctdb_req_message_old, data) + buflen;
 607         r.hdr.ctdb_magic = CTDB_MAGIC;
 608         r.hdr.ctdb_version = CTDB_PROTOCOL;
 609         r.hdr.generation = 1;
 610         r.hdr.operation  = CTDB_REQ_MESSAGE;
 611         r.hdr.destnode   = dst_vnn;
 612         r.hdr.srcnode    = conn->our_vnn;
 613         r.hdr.reqid      = 0;
 614         r.srvid          = dst_srvid;
 615         r.datalen        = buflen;
 616
 617         DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
 618         ctdb_packet_dump(&r.hdr);
 619
 620         iov2[0].iov_base = &r;
 621         iov2[0].iov_len = offsetof(struct ctdb_req_message_old, data);
 622         memcpy(&iov2[1], iov, iovlen * sizeof(struct iovec));
 623
 624         nwritten = write_data_iov(conn->fd, iov2, iovlen+1);
 625         if (nwritten == -1) {
 626                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
 627                 cluster_fatal("cluster dispatch daemon msg write error\n");
 628         }
 629
 630         return 0;
 631 }
 632
 633 /*
 634  * send/recv a generic ctdb control message
 635  */
 636 static int ctdbd_control(struct ctdbd_connection *conn,
 637                          uint32_t vnn, uint32_t opcode,
 638                          uint64_t srvid, uint32_t flags,
 639                          TDB_DATA data,
 640                          TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
 641                          int *cstatus)
 642 {
 643         struct ctdb_req_control_old req;
 644         struct ctdb_req_header *hdr;
 645         struct ctdb_reply_control_old *reply = NULL;
 646         struct iovec iov[2];
 647         ssize_t nwritten;
 648         int ret;
 649
 650         ZERO_STRUCT(req);
 651         req.hdr.length = offsetof(struct ctdb_req_control_old, data) + data.dsize;
 652         req.hdr.ctdb_magic   = CTDB_MAGIC;
 653         req.hdr.ctdb_version = CTDB_PROTOCOL;
 654         req.hdr.operation    = CTDB_REQ_CONTROL;
 655         req.hdr.reqid        = ctdbd_next_reqid(conn);
 656         req.hdr.destnode     = vnn;
 657         req.opcode           = opcode;
 658         req.srvid            = srvid;
 659         req.datalen          = data.dsize;
 660         req.flags            = flags;
 661
 662         DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
 663         ctdb_packet_dump(&req.hdr);
 664
 665         iov[0].iov_base = &req;
 666         iov[0].iov_len = offsetof(struct ctdb_req_control_old, data);
 667         iov[1].iov_base = data.dptr;
 668         iov[1].iov_len = data.dsize;
 669
 670         nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
 671         if (nwritten == -1) {
 672                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
 673                 cluster_fatal("cluster dispatch daemon msg write error\n");
 674         }
 675
 676         if (flags & CTDB_CTRL_FLAG_NOREPLY) {
 677                 if (cstatus) {
 678                         *cstatus = 0;
 679                 }
 680                 return 0;
 681         }
 682
 683         ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
 684         if (ret != 0) {
 685                 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
 686                 return ret;
 687         }
 688
 689         if (hdr->operation != CTDB_REPLY_CONTROL) {
 690                 DEBUG(0, ("received invalid reply\n"));
 691                 TALLOC_FREE(hdr);
 692                 return EIO;
 693         }
 694         reply = (struct ctdb_reply_control_old *)hdr;
 695
 696         if (outdata) {
 697                 if (!(outdata->dptr = (uint8_t *)talloc_memdup(
 698                               mem_ctx, reply->data, reply->datalen))) {
 699                         TALLOC_FREE(reply);
 700                         return ENOMEM;
 701                 }
 702                 outdata->dsize = reply->datalen;
 703         }
 704         if (cstatus) {
 705                 (*cstatus) = reply->status;
 706         }
 707
 708         TALLOC_FREE(reply);
 709         return ret;
 710 }
 711
 712 /*
 713  * see if a remote process exists
 714  */
 715 bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32_t vnn, pid_t pid)
 716 {
 717         struct server_id id;
 718         bool result;
 719
 720         id.pid = pid;
 721         id.vnn = vnn;
 722
 723         if (!ctdb_processes_exist(conn, &id, 1, &result)) {
 724                 DEBUG(10, ("ctdb_processes_exist failed\n"));
 725                 return false;
 726         }
 727         return result;
 728 }
 729
 730 bool ctdb_processes_exist(struct ctdbd_connection *conn,
 731                           const struct server_id *pids, int num_pids,
 732                           bool *results)
 733 {
 734         TALLOC_CTX *frame = talloc_stackframe();
 735         int i, num_received;
 736         uint32_t *reqids;
 737         bool result = false;
 738
 739         reqids = talloc_array(talloc_tos(), uint32_t, num_pids);
 740         if (reqids == NULL) {
 741                 goto fail;
 742         }
 743
 744         for (i=0; i<num_pids; i++) {
 745                 struct ctdb_req_control_old req;
 746                 pid_t pid;
 747                 struct iovec iov[2];
 748                 ssize_t nwritten;
 749
 750                 results[i] = false;
 751                 reqids[i] = ctdbd_next_reqid(conn);
 752
 753                 ZERO_STRUCT(req);
 754
 755                 /*
 756                  * pids[i].pid is uint64_t, scale down to pid_t which
 757                  * is the wire protocol towards ctdb.
 758                  */
 759                 pid = pids[i].pid;
 760
 761                 DEBUG(10, ("Requesting PID %d/%d, reqid=%d\n",
 762                            (int)pids[i].vnn, (int)pid,
 763                            (int)reqids[i]));
 764
 765                 req.hdr.length = offsetof(struct ctdb_req_control_old, data);
 766                 req.hdr.length += sizeof(pid);
 767                 req.hdr.ctdb_magic   = CTDB_MAGIC;
 768                 req.hdr.ctdb_version = CTDB_PROTOCOL;
 769                 req.hdr.operation    = CTDB_REQ_CONTROL;
 770                 req.hdr.reqid        = reqids[i];
 771                 req.hdr.destnode     = pids[i].vnn;
 772                 req.opcode           = CTDB_CONTROL_PROCESS_EXISTS;
 773                 req.srvid            = 0;
 774                 req.datalen          = sizeof(pid);
 775                 req.flags            = 0;
 776
 777                 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
 778                 ctdb_packet_dump(&req.hdr);
 779
 780                 iov[0].iov_base = &req;
 781                 iov[0].iov_len = offsetof(struct ctdb_req_control_old, data);
 782                 iov[1].iov_base = &pid;
 783                 iov[1].iov_len = sizeof(pid);
 784
 785                 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
 786                 if (nwritten == -1) {
 787                         DEBUG(10, ("write_data_iov failed: %s\n",
 788                                    strerror(errno)));
 789                         goto fail;
 790                 }
 791         }
 792
 793         num_received = 0;
 794
 795         while (num_received < num_pids) {
 796                 struct ctdb_req_header *hdr;
 797                 struct ctdb_reply_control_old *reply;
 798                 uint32_t reqid;
 799                 int ret;
 800
 801                 ret = ctdb_read_req(conn, 0, talloc_tos(), &hdr);
 802                 if (ret != 0) {
 803                         DEBUG(10, ("ctdb_read_req failed: %s\n",
 804                                    strerror(ret)));
 805                         goto fail;
 806                 }
 807
 808                 if (hdr->operation != CTDB_REPLY_CONTROL) {
 809                         DEBUG(10, ("Received invalid reply\n"));
 810                         goto fail;
 811                 }
 812                 reply = (struct ctdb_reply_control_old *)hdr;
 813
 814                 reqid = reply->hdr.reqid;
 815
 816                 DEBUG(10, ("Received reqid %d\n", (int)reqid));
 817
 818                 for (i=0; i<num_pids; i++) {
 819                         if (reqid == reqids[i]) {
 820                                 break;
 821                         }
 822                 }
 823                 if (i == num_pids) {
 824                         DEBUG(10, ("Received unknown record number %u\n",
 825                                    (unsigned)reqid));
 826                         goto fail;
 827                 }
 828                 results[i] = ((reply->status) == 0);
 829                 TALLOC_FREE(reply);
 830                 num_received += 1;
 831         }
 832
 833         result = true;
 834 fail:
 835         TALLOC_FREE(frame);
 836         return result;
 837 }
 838
 839 /*
 840  * Get a db path
 841  */
 842 char *ctdbd_dbpath(struct ctdbd_connection *conn,
 843                    TALLOC_CTX *mem_ctx, uint32_t db_id)
 844 {
 845         int ret;
 846         TDB_DATA data;
 847         TDB_DATA rdata = {0};
 848         int32_t cstatus = 0;
 849
 850         data.dptr = (uint8_t*)&db_id;
 851         data.dsize = sizeof(db_id);
 852
 853         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
 854                             CTDB_CONTROL_GETDBPATH, 0, 0, data,
 855                             mem_ctx, &rdata, &cstatus);
 856         if ((ret != 0) || cstatus != 0) {
 857                 DEBUG(0, (__location__ " ctdb_control for getdbpath failed: %s\n",
 858                           strerror(ret)));
 859                 return NULL;
 860         }
 861
 862         return (char *)rdata.dptr;
 863 }
 864
 865 /*
 866  * attach to a ctdb database
 867  */
 868 int ctdbd_db_attach(struct ctdbd_connection *conn,
 869                     const char *name, uint32_t *db_id, int tdb_flags)
 870 {
 871         int ret;
 872         TDB_DATA data;
 873         int32_t cstatus;
 874         bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
 875
 876         data = string_term_tdb_data(name);
 877
 878         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
 879                             persistent
 880                             ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
 881                             : CTDB_CONTROL_DB_ATTACH,
 882                             tdb_flags, 0, data, NULL, &data, &cstatus);
 883         if (ret != 0) {
 884                 DEBUG(0, (__location__ " ctdb_control for db_attach "
 885                           "failed: %s\n", strerror(ret)));
 886                 return ret;
 887         }
 888
 889         if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
 890                 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
 891                 return EIO;
 892         }
 893
 894         *db_id = *(uint32_t *)data.dptr;
 895         talloc_free(data.dptr);
 896
 897         if (!(tdb_flags & TDB_SEQNUM)) {
 898                 return 0;
 899         }
 900
 901         data.dptr = (uint8_t *)db_id;
 902         data.dsize = sizeof(*db_id);
 903
 904         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
 905                             CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
 906                             NULL, NULL, &cstatus);
 907         if ((ret != 0) || cstatus != 0) {
 908                 DEBUG(0, (__location__ " ctdb_control for enable seqnum "
 909                           "failed: %s\n", strerror(ret)));
 910                 return (ret == 0) ? EIO : ret;
 911         }
 912
 913         return 0;
 914 }
 915
 916 /*
 917  * force the migration of a record to this node
 918  */
 919 int ctdbd_migrate(struct ctdbd_connection *conn, uint32_t db_id, TDB_DATA key)
 920 {
 921         struct ctdb_req_call_old req;
 922         struct ctdb_req_header *hdr;
 923         struct iovec iov[2];
 924         ssize_t nwritten;
 925         int ret;
 926
 927         ZERO_STRUCT(req);
 928
 929         req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
 930         req.hdr.ctdb_magic   = CTDB_MAGIC;
 931         req.hdr.ctdb_version = CTDB_PROTOCOL;
 932         req.hdr.operation    = CTDB_REQ_CALL;
 933         req.hdr.reqid        = ctdbd_next_reqid(conn);
 934         req.flags            = CTDB_IMMEDIATE_MIGRATION;
 935         req.callid           = CTDB_NULL_FUNC;
 936         req.db_id            = db_id;
 937         req.keylen           = key.dsize;
 938
 939         DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
 940         ctdb_packet_dump(&req.hdr);
 941
 942         iov[0].iov_base = &req;
 943         iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
 944         iov[1].iov_base = key.dptr;
 945         iov[1].iov_len = key.dsize;
 946
 947         nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
 948         if (nwritten == -1) {
 949                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
 950                 cluster_fatal("cluster dispatch daemon msg write error\n");
 951         }
 952
 953         ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
 954         if (ret != 0) {
 955                 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
 956                 goto fail;
 957         }
 958
 959         if (hdr->operation != CTDB_REPLY_CALL) {
 960                 DEBUG(0, ("received invalid reply\n"));
 961                 goto fail;
 962         }
 963
 964  fail:
 965
 966         TALLOC_FREE(hdr);
 967         return ret;
 968 }
 969
 970 /*
 971  * Fetch a record and parse it
 972  */
 973 int ctdbd_parse(struct ctdbd_connection *conn, uint32_t db_id,
 974                 TDB_DATA key, bool local_copy,
 975                 void (*parser)(TDB_DATA key, TDB_DATA data,
 976                                void *private_data),
 977                 void *private_data)
 978 {
 979         struct ctdb_req_call_old req;
 980         struct ctdb_req_header *hdr = NULL;
 981         struct ctdb_reply_call_old *reply;
 982         struct iovec iov[2];
 983         ssize_t nwritten;
 984         uint32_t flags;
 985         int ret;
 986
 987         flags = local_copy ? CTDB_WANT_READONLY : 0;
 988
 989         ZERO_STRUCT(req);
 990
 991         req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
 992         req.hdr.ctdb_magic   = CTDB_MAGIC;
 993         req.hdr.ctdb_version = CTDB_PROTOCOL;
 994         req.hdr.operation    = CTDB_REQ_CALL;
 995         req.hdr.reqid        = ctdbd_next_reqid(conn);
 996         req.flags            = flags;
 997         req.callid           = CTDB_FETCH_FUNC;
 998         req.db_id            = db_id;
 999         req.keylen           = key.dsize;
1000
1001         iov[0].iov_base = &req;
1002         iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
1003         iov[1].iov_base = key.dptr;
1004         iov[1].iov_len = key.dsize;
1005
1006         nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
1007         if (nwritten == -1) {
1008                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
1009                 cluster_fatal("cluster dispatch daemon msg write error\n");
1010         }
1011
1012         ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
1013         if (ret != 0) {
1014                 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
1015                 goto fail;
1016         }
1017
1018         if ((hdr == NULL) || (hdr->operation != CTDB_REPLY_CALL)) {
1019                 DEBUG(0, ("received invalid reply\n"));
1020                 ret = EIO;
1021                 goto fail;
1022         }
1023         reply = (struct ctdb_reply_call_old *)hdr;
1024
1025         if (reply->datalen == 0) {
1026                 /*
1027                  * Treat an empty record as non-existing
1028                  */
1029                 ret = ENOENT;
1030                 goto fail;
1031         }
1032
1033         parser(key, make_tdb_data(&reply->data[0], reply->datalen),
1034                private_data);
1035
1036         ret = 0;
1037  fail:
1038         TALLOC_FREE(hdr);
1039         return ret;
1040 }
1041
1042 /*
1043   Traverse a ctdb database. This uses a kind-of hackish way to open a second
1044   connection to ctdbd to avoid the hairy recursive and async problems with
1045   everything in-line.
1046 */
1047
1048 int ctdbd_traverse(struct ctdbd_connection *master, uint32_t db_id,
1049                         void (*fn)(TDB_DATA key, TDB_DATA data,
1050                                    void *private_data),
1051                         void *private_data)
1052 {
1053         struct ctdbd_connection *conn;
1054         int ret;
1055         TDB_DATA key, data;
1056         struct ctdb_traverse_start t;
1057         int cstatus;
1058
1059         become_root();
1060         ret = ctdbd_init_connection(NULL, master->sockname, master->timeout,
1061                                     &conn);
1062         unbecome_root();
1063         if (ret != 0) {
1064                 DEBUG(0, ("ctdbd_init_connection failed: %s\n",
1065                           strerror(ret)));
1066                 return ret;
1067         }
1068
1069         t.db_id = db_id;
1070         t.srvid = conn->rand_srvid;
1071         t.reqid = ctdbd_next_reqid(conn);
1072
1073         data.dptr = (uint8_t *)&t;
1074         data.dsize = sizeof(t);
1075
1076         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
1077                             CTDB_CONTROL_TRAVERSE_START, conn->rand_srvid,
1078                             0, data, NULL, NULL, &cstatus);
1079
1080         if ((ret != 0) || (cstatus != 0)) {
1081                 DEBUG(0,("ctdbd_control failed: %s, %d\n", strerror(ret),
1082                          cstatus));
1083
1084                 if (ret == 0) {
1085                         /*
1086                          * We need a mapping here
1087                          */
1088                         ret = EIO;
1089                 }
1090                 TALLOC_FREE(conn);
1091                 return ret;
1092         }
1093
1094         while (True) {
1095                 struct ctdb_req_header *hdr = NULL;
1096                 struct ctdb_req_message_old *m;
1097                 struct ctdb_rec_data_old *d;
1098
1099                 ret = ctdb_read_packet(conn->fd, conn->timeout, conn, &hdr);
1100                 if (ret != 0) {
1101                         DEBUG(0, ("ctdb_read_packet failed: %s\n",
1102                                   strerror(ret)));
1103                         cluster_fatal("ctdbd died\n");
1104                 }
1105
1106                 if (hdr->operation != CTDB_REQ_MESSAGE) {
1107                         DEBUG(0, ("Got operation %u, expected a message\n",
1108                                   (unsigned)hdr->operation));
1109                         TALLOC_FREE(conn);
1110                         return EIO;
1111                 }
1112
1113                 m = (struct ctdb_req_message_old *)hdr;
1114                 d = (struct ctdb_rec_data_old *)&m->data[0];
1115                 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1116                         DEBUG(0, ("Got invalid traverse data of length %d\n",
1117                                   (int)m->datalen));
1118                         TALLOC_FREE(conn);
1119                         return EIO;
1120                 }
1121
1122                 key.dsize = d->keylen;
1123                 key.dptr  = &d->data[0];
1124                 data.dsize = d->datalen;
1125                 data.dptr = &d->data[d->keylen];
1126
1127                 if (key.dsize == 0 && data.dsize == 0) {
1128                         /* end of traverse */
1129                         TALLOC_FREE(conn);
1130                         return 0;
1131                 }
1132
1133                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1134                         DEBUG(0, ("Got invalid ltdb header length %d\n",
1135                                   (int)data.dsize));
1136                         TALLOC_FREE(conn);
1137                         return EIO;
1138                 }
1139                 data.dsize -= sizeof(struct ctdb_ltdb_header);
1140                 data.dptr += sizeof(struct ctdb_ltdb_header);
1141
1142                 if (fn != NULL) {
1143                         fn(key, data, private_data);
1144                 }
1145         }
1146         return 0;
1147 }
1148
1149 /*
1150    This is used to canonicalize a ctdb_sock_addr structure.
1151 */
1152 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1153                                       struct sockaddr_storage *out)
1154 {
1155         memcpy(out, in, sizeof (*out));
1156
1157 #ifdef HAVE_IPV6
1158         if (in->ss_family == AF_INET6) {
1159                 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1160                 const struct sockaddr_in6 *in6 =
1161                         (const struct sockaddr_in6 *)in;
1162                 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1163                 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1164                         memset(out, 0, sizeof(*out));
1165 #ifdef HAVE_SOCK_SIN_LEN
1166                         out4->sin_len = sizeof(*out);
1167 #endif
1168                         out4->sin_family = AF_INET;
1169                         out4->sin_port   = in6->sin6_port;
1170                         memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr[12], 4);
1171                 }
1172         }
1173 #endif
1174 }
1175
1176 /*
1177  * Register us as a server for a particular tcp connection
1178  */
1179
1180 int ctdbd_register_ips(struct ctdbd_connection *conn,
1181                        const struct sockaddr_storage *_server,
1182                        const struct sockaddr_storage *_client,
1183                        int (*cb)(uint32_t src_vnn, uint32_t dst_vnn,
1184                                  uint64_t dst_srvid,
1185                                  const uint8_t *msg, size_t msglen,
1186                                  void *private_data),
1187                        void *private_data)
1188 {
1189         struct ctdb_control_tcp_addr p;
1190         TDB_DATA data = { .dptr = (uint8_t *)&p, .dsize = sizeof(p) };
1191         int ret;
1192         struct sockaddr_storage client;
1193         struct sockaddr_storage server;
1194
1195         /*
1196          * Only one connection so far
1197          */
1198
1199         smbd_ctdb_canonicalize_ip(_client, &client);
1200         smbd_ctdb_canonicalize_ip(_server, &server);
1201
1202         switch (client.ss_family) {
1203         case AF_INET:
1204                 memcpy(&p.dest.ip, &server, sizeof(p.dest.ip));
1205                 memcpy(&p.src.ip, &client, sizeof(p.src.ip));
1206                 break;
1207         case AF_INET6:
1208                 memcpy(&p.dest.ip6, &server, sizeof(p.dest.ip6));
1209                 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1210                 break;
1211         default:
1212                 return EIO;
1213         }
1214
1215         /*
1216          * We want to be told about IP releases
1217          */
1218
1219         ret = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP,
1220                                   cb, private_data);
1221         if (ret != 0) {
1222                 return ret;
1223         }
1224
1225         /*
1226          * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1227          * can send an extra ack to trigger a reset for our client, so it
1228          * immediately reconnects
1229          */
1230         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
1231                             CTDB_CONTROL_TCP_CLIENT, 0,
1232                             CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL,
1233                             NULL);
1234         if (ret != 0) {
1235                 return ret;
1236         }
1237         return 0;
1238 }
1239
1240 /*
1241   call a control on the local node
1242  */
1243 int ctdbd_control_local(struct ctdbd_connection *conn, uint32_t opcode,
1244                         uint64_t srvid, uint32_t flags, TDB_DATA data,
1245                         TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1246                         int *cstatus)
1247 {
1248         return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data,
1249                              mem_ctx, outdata, cstatus);
1250 }
1251
1252 int ctdb_watch_us(struct ctdbd_connection *conn)
1253 {
1254         struct ctdb_client_notify_register reg_data;
1255         size_t struct_len;
1256         int ret;
1257         int cstatus;
1258
1259         reg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1260         reg_data.len = 1;
1261         reg_data.notify_data[0] = 0;
1262
1263         struct_len = offsetof(struct ctdb_client_notify_register,
1264                               notify_data) + reg_data.len;
1265
1266         ret = ctdbd_control_local(
1267                 conn, CTDB_CONTROL_REGISTER_NOTIFY, conn->rand_srvid, 0,
1268                 make_tdb_data((uint8_t *)&reg_data, struct_len),
1269                 NULL, NULL, &cstatus);
1270         if (ret != 0) {
1271                 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1272                           strerror(ret)));
1273         }
1274         return ret;
1275 }
1276
1277 int ctdb_unwatch(struct ctdbd_connection *conn)
1278 {
1279         struct ctdb_client_notify_deregister dereg_data;
1280         int ret;
1281         int cstatus;
1282
1283         dereg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1284
1285         ret = ctdbd_control_local(
1286                 conn, CTDB_CONTROL_DEREGISTER_NOTIFY, conn->rand_srvid, 0,
1287                 make_tdb_data((uint8_t *)&dereg_data, sizeof(dereg_data)),
1288                 NULL, NULL, &cstatus);
1289         if (ret != 0) {
1290                 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1291                           strerror(ret)));
1292         }
1293         return ret;
1294 }
1295
1296 int ctdbd_probe(const char *sockname, int timeout)
1297 {
1298         /*
1299          * Do a very early check if ctdbd is around to avoid an abort and core
1300          * later
1301          */
1302         struct ctdbd_connection *conn = NULL;
1303         int ret;
1304
1305         ret = ctdbd_messaging_connection(talloc_tos(), sockname, timeout,
1306                                          &conn);
1307
1308         /*
1309          * We only care if we can connect.
1310          */
1311         TALLOC_FREE(conn);
1312
1313         return ret;
1314 }