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 "replace.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 #include "lib/util/select.h"
  29 #include "lib/util/debug.h"
  30 #include "lib/util/talloc_stack.h"
  31 #include "lib/util/genrand.h"
  32 #include "lib/util/fault.h"
  33
  34 /* paths to these include files come from --with-ctdb= in configure */
  35
  36 #include "ctdb_private.h"
  37
  38 struct ctdbd_srvid_cb {
  39         uint64_t srvid;
  40         int (*cb)(uint32_t src_vnn, uint32_t dst_vnn,
  41                   uint64_t dst_srvid,
  42                   const uint8_t *msg, size_t msglen,
  43                   void *private_data);
  44         void *private_data;
  45 };
  46
  47 struct ctdbd_connection {
  48         uint32_t reqid;
  49         uint32_t our_vnn;
  50         uint64_t rand_srvid;
  51         struct ctdbd_srvid_cb *callbacks;
  52         int fd;
  53         struct tevent_fd *fde;
  54         int timeout;
  55 };
  56
  57 static uint32_t ctdbd_next_reqid(struct ctdbd_connection *conn)
  58 {
  59         conn->reqid += 1;
  60         if (conn->reqid == 0) {
  61                 conn->reqid += 1;
  62         }
  63         return conn->reqid;
  64 }
  65
  66 static int ctdbd_control(struct ctdbd_connection *conn,
  67                          uint32_t vnn, uint32_t opcode,
  68                          uint64_t srvid, uint32_t flags,
  69                          TDB_DATA data,
  70                          TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
  71                          int32_t *cstatus);
  72
  73 /*
  74  * exit on fatal communications errors with the ctdbd daemon
  75  */
  76 static void cluster_fatal(const char *why)
  77 {
  78         DEBUG(0,("cluster fatal event: %s - exiting immediately\n", why));
  79         /* we don't use smb_panic() as we don't want to delay to write
  80            a core file. We need to release this process id immediately
  81            so that someone else can take over without getting sharing
  82            violations */
  83         _exit(1);
  84 }
  85
  86 /*
  87  *
  88  */
  89 static void ctdb_packet_dump(struct ctdb_req_header *hdr)
  90 {
  91         if (DEBUGLEVEL < 11) {
  92                 return;
  93         }
  94         DEBUGADD(11, ("len=%d, magic=%x, vers=%d, gen=%d, op=%d, reqid=%d\n",
  95                       (int)hdr->length, (int)hdr->ctdb_magic,
  96                       (int)hdr->ctdb_version, (int)hdr->generation,
  97                       (int)hdr->operation, (int)hdr->reqid));
  98 }
  99
 100 /*
 101  * Register a srvid with ctdbd
 102  */
 103 int register_with_ctdbd(struct ctdbd_connection *conn, uint64_t srvid,
 104                         int (*cb)(uint32_t src_vnn, uint32_t dst_vnn,
 105                                   uint64_t dst_srvid,
 106                                   const uint8_t *msg, size_t msglen,
 107                                   void *private_data),
 108                         void *private_data)
 109 {
 110
 111         int ret;
 112         int32_t cstatus;
 113         size_t num_callbacks;
 114         struct ctdbd_srvid_cb *tmp;
 115
 116         ret = ctdbd_control_local(conn, CTDB_CONTROL_REGISTER_SRVID, srvid, 0,
 117                                   tdb_null, NULL, NULL, &cstatus);
 118         if (ret != 0) {
 119                 return ret;
 120         }
 121
 122         num_callbacks = talloc_array_length(conn->callbacks);
 123
 124         tmp = talloc_realloc(conn, conn->callbacks, struct ctdbd_srvid_cb,
 125                              num_callbacks + 1);
 126         if (tmp == NULL) {
 127                 return ENOMEM;
 128         }
 129         conn->callbacks = tmp;
 130
 131         conn->callbacks[num_callbacks] = (struct ctdbd_srvid_cb) {
 132                 .srvid = srvid, .cb = cb, .private_data = private_data
 133         };
 134
 135         return 0;
 136 }
 137
 138 static int ctdbd_msg_call_back(struct ctdbd_connection *conn,
 139                                struct ctdb_req_message_old *msg)
 140 {
 141         uint32_t msg_len;
 142         size_t i, num_callbacks;
 143
 144         msg_len = msg->hdr.length;
 145         if (msg_len < offsetof(struct ctdb_req_message_old, data)) {
 146                 DBG_DEBUG("len %"PRIu32" too small\n", msg_len);
 147                 return 0;
 148         }
 149         msg_len -= offsetof(struct ctdb_req_message_old, data);
 150
 151         if (msg_len < msg->datalen) {
 152                 DBG_DEBUG("msg_len=%"PRIu32" < msg->datalen=%"PRIu32"\n",
 153                           msg_len, 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_local(conn, CTDB_CONTROL_GET_PNN, 0, 0,
 184                                   tdb_null, NULL, NULL, &cstatus);
 185         if (ret != 0) {
 186                 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
 187                 return ret;
 188         }
 189         *vnn = (uint32_t)cstatus;
 190         return ret;
 191 }
 192
 193 /*
 194  * Are we active (i.e. not banned or stopped?)
 195  */
 196 static bool ctdbd_working(struct ctdbd_connection *conn, uint32_t vnn)
 197 {
 198         int32_t cstatus=-1;
 199         TDB_DATA outdata;
 200         struct ctdb_node_map_old *m;
 201         uint32_t failure_flags;
 202         bool ok = false;
 203         uint32_t i;
 204         int ret;
 205
 206         ret = ctdbd_control_local(conn, CTDB_CONTROL_GET_NODEMAP, 0, 0,
 207                                   tdb_null, talloc_tos(), &outdata, &cstatus);
 208         if (ret != 0) {
 209                 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
 210                 return false;
 211         }
 212         if ((cstatus != 0) || (outdata.dptr == NULL)) {
 213                 DEBUG(2, ("Received invalid ctdb data\n"));
 214                 return false;
 215         }
 216
 217         m = (struct ctdb_node_map_old *)outdata.dptr;
 218
 219         for (i=0; i<m->num; i++) {
 220                 if (vnn == m->nodes[i].pnn) {
 221                         break;
 222                 }
 223         }
 224
 225         if (i == m->num) {
 226                 DEBUG(2, ("Did not find ourselves (node %d) in nodemap\n",
 227                           (int)vnn));
 228                 goto fail;
 229         }
 230
 231         failure_flags = NODE_FLAGS_BANNED | NODE_FLAGS_DISCONNECTED
 232                 | NODE_FLAGS_PERMANENTLY_DISABLED | NODE_FLAGS_STOPPED;
 233
 234         if ((m->nodes[i].flags & failure_flags) != 0) {
 235                 DEBUG(2, ("Node has status %x, not active\n",
 236                           (int)m->nodes[i].flags));
 237                 goto fail;
 238         }
 239
 240         ok = true;
 241 fail:
 242         TALLOC_FREE(outdata.dptr);
 243         return ok;
 244 }
 245
 246 uint32_t ctdbd_vnn(const struct ctdbd_connection *conn)
 247 {
 248         return conn->our_vnn;
 249 }
 250
 251 /*
 252  * Get us a ctdb connection
 253  */
 254
 255 static int ctdbd_connect(const char *sockname, int *pfd)
 256 {
 257         struct sockaddr_un addr = { 0, };
 258         int fd;
 259         socklen_t salen;
 260         size_t namelen;
 261
 262         fd = socket(AF_UNIX, SOCK_STREAM, 0);
 263         if (fd == -1) {
 264                 int err = errno;
 265                 DEBUG(3, ("Could not create socket: %s\n", strerror(err)));
 266                 return err;
 267         }
 268
 269         addr.sun_family = AF_UNIX;
 270
 271         namelen = strlcpy(addr.sun_path, sockname, sizeof(addr.sun_path));
 272         if (namelen >= sizeof(addr.sun_path)) {
 273                 DEBUG(3, ("%s: Socket name too long: %s\n", __func__,
 274                           sockname));
 275                 close(fd);
 276                 return ENAMETOOLONG;
 277         }
 278
 279         salen = sizeof(struct sockaddr_un);
 280
 281         if (connect(fd, (struct sockaddr *)(void *)&addr, salen) == -1) {
 282                 int err = errno;
 283                 DEBUG(1, ("connect(%s) failed: %s\n", sockname,
 284                           strerror(err)));
 285                 close(fd);
 286                 return err;
 287         }
 288
 289         *pfd = fd;
 290         return 0;
 291 }
 292
 293 static int ctdb_read_packet(int fd, int timeout, TALLOC_CTX *mem_ctx,
 294                             struct ctdb_req_header **result)
 295 {
 296         struct ctdb_req_header *req;
 297         uint32_t msglen;
 298         ssize_t nread;
 299
 300         if (timeout != -1) {
 301                 struct pollfd pfd = { .fd = fd, .events = POLLIN };
 302                 int ret;
 303
 304                 ret = sys_poll_intr(&pfd, 1, timeout);
 305                 if (ret == -1) {
 306                         return errno;
 307                 }
 308                 if (ret == 0) {
 309                         return ETIMEDOUT;
 310                 }
 311                 if (ret != 1) {
 312                         return EIO;
 313                 }
 314         }
 315
 316         nread = read_data(fd, &msglen, sizeof(msglen));
 317         if (nread == -1) {
 318                 return errno;
 319         }
 320         if (nread == 0) {
 321                 return EIO;
 322         }
 323
 324         if (msglen < sizeof(struct ctdb_req_header)) {
 325                 return EIO;
 326         }
 327
 328         req = talloc_size(mem_ctx, msglen);
 329         if (req == NULL) {
 330                 return ENOMEM;
 331         }
 332         talloc_set_name_const(req, "struct ctdb_req_header");
 333
 334         req->length = msglen;
 335
 336         nread = read_data(fd, ((char *)req) + sizeof(msglen),
 337                           msglen - sizeof(msglen));
 338         if (nread == -1) {
 339                 TALLOC_FREE(req);
 340                 return errno;
 341         }
 342         if (nread == 0) {
 343                 TALLOC_FREE(req);
 344                 return EIO;
 345         }
 346
 347         *result = req;
 348         return 0;
 349 }
 350
 351 /*
 352  * Read a full ctdbd request. If we have a messaging context, defer incoming
 353  * messages that might come in between.
 354  */
 355
 356 static int ctdb_read_req(struct ctdbd_connection *conn, uint32_t reqid,
 357                          TALLOC_CTX *mem_ctx, struct ctdb_req_header **result)
 358 {
 359         struct ctdb_req_header *hdr;
 360         int ret;
 361
 362  next_pkt:
 363
 364         ret = ctdb_read_packet(conn->fd, conn->timeout, mem_ctx, &hdr);
 365         if (ret != 0) {
 366                 DEBUG(0, ("ctdb_read_packet failed: %s\n", strerror(ret)));
 367                 cluster_fatal("ctdbd died\n");
 368         }
 369
 370         DEBUG(11, ("Received ctdb packet\n"));
 371         ctdb_packet_dump(hdr);
 372
 373         if (hdr->operation == CTDB_REQ_MESSAGE) {
 374                 struct ctdb_req_message_old *msg = (struct ctdb_req_message_old *)hdr;
 375
 376                 ret = ctdbd_msg_call_back(conn, msg);
 377                 if (ret != 0) {
 378                         TALLOC_FREE(hdr);
 379                         return ret;
 380                 }
 381
 382                 TALLOC_FREE(hdr);
 383                 goto next_pkt;
 384         }
 385
 386         if ((reqid != 0) && (hdr->reqid != reqid)) {
 387                 /* we got the wrong reply */
 388                 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
 389                          "been %u\n", hdr->reqid, reqid));
 390                 TALLOC_FREE(hdr);
 391                 goto next_pkt;
 392         }
 393
 394         *result = talloc_move(mem_ctx, &hdr);
 395
 396         return 0;
 397 }
 398
 399 static int ctdbd_connection_destructor(struct ctdbd_connection *c)
 400 {
 401         TALLOC_FREE(c->fde);
 402         if (c->fd != -1) {
 403                 close(c->fd);
 404                 c->fd = -1;
 405         }
 406         return 0;
 407 }
 408 /*
 409  * Get us a ctdbd connection
 410  */
 411
 412 int ctdbd_init_connection(TALLOC_CTX *mem_ctx,
 413                           const char *sockname, int timeout,
 414                           struct ctdbd_connection **pconn)
 415 {
 416         struct ctdbd_connection *conn;
 417         int ret;
 418
 419         if (!(conn = talloc_zero(mem_ctx, struct ctdbd_connection))) {
 420                 DEBUG(0, ("talloc failed\n"));
 421                 return ENOMEM;
 422         }
 423
 424         conn->timeout = timeout;
 425
 426         if (conn->timeout == 0) {
 427                 conn->timeout = -1;
 428         }
 429
 430         ret = ctdbd_connect(sockname, &conn->fd);
 431         if (ret != 0) {
 432                 DEBUG(1, ("ctdbd_connect failed: %s\n", strerror(ret)));
 433                 goto fail;
 434         }
 435         talloc_set_destructor(conn, ctdbd_connection_destructor);
 436
 437         ret = get_cluster_vnn(conn, &conn->our_vnn);
 438
 439         if (ret != 0) {
 440                 DEBUG(10, ("get_cluster_vnn failed: %s\n", strerror(ret)));
 441                 goto fail;
 442         }
 443
 444         if (!ctdbd_working(conn, conn->our_vnn)) {
 445                 DEBUG(2, ("Node is not working, can not connect\n"));
 446                 ret = EIO;
 447                 goto fail;
 448         }
 449
 450         generate_random_buffer((unsigned char *)&conn->rand_srvid,
 451                                sizeof(conn->rand_srvid));
 452
 453         ret = register_with_ctdbd(conn, conn->rand_srvid, NULL, NULL);
 454
 455         if (ret != 0) {
 456                 DEBUG(5, ("Could not register random srvid: %s\n",
 457                           strerror(ret)));
 458                 goto fail;
 459         }
 460
 461         *pconn = conn;
 462         return 0;
 463
 464  fail:
 465         TALLOC_FREE(conn);
 466         return ret;
 467 }
 468
 469 int ctdbd_conn_get_fd(struct ctdbd_connection *conn)
 470 {
 471         return conn->fd;
 472 }
 473
 474 /*
 475  * Packet handler to receive and handle a ctdb message
 476  */
 477 static int ctdb_handle_message(struct ctdbd_connection *conn,
 478                                struct ctdb_req_header *hdr)
 479 {
 480         struct ctdb_req_message_old *msg;
 481
 482         if (hdr->operation != CTDB_REQ_MESSAGE) {
 483                 DEBUG(0, ("Received async msg of type %u, discarding\n",
 484                           hdr->operation));
 485                 return EINVAL;
 486         }
 487
 488         msg = (struct ctdb_req_message_old *)hdr;
 489
 490         ctdbd_msg_call_back(conn, msg);
 491
 492         return 0;
 493 }
 494
 495 void ctdbd_socket_readable(struct ctdbd_connection *conn)
 496 {
 497         struct ctdb_req_header *hdr = NULL;
 498         int ret;
 499
 500         ret = ctdb_read_packet(conn->fd, conn->timeout, talloc_tos(), &hdr);
 501         if (ret != 0) {
 502                 DEBUG(0, ("ctdb_read_packet failed: %s\n", strerror(ret)));
 503                 cluster_fatal("ctdbd died\n");
 504         }
 505
 506         ret = ctdb_handle_message(conn, hdr);
 507
 508         TALLOC_FREE(hdr);
 509
 510         if (ret != 0) {
 511                 DEBUG(10, ("could not handle incoming message: %s\n",
 512                            strerror(ret)));
 513         }
 514 }
 515
 516 int ctdbd_messaging_send_iov(struct ctdbd_connection *conn,
 517                              uint32_t dst_vnn, uint64_t dst_srvid,
 518                              const struct iovec *iov, int iovlen)
 519 {
 520         struct ctdb_req_message_old r;
 521         struct iovec iov2[iovlen+1];
 522         size_t buflen = iov_buflen(iov, iovlen);
 523         ssize_t nwritten;
 524
 525         r.hdr.length = offsetof(struct ctdb_req_message_old, data) + buflen;
 526         r.hdr.ctdb_magic = CTDB_MAGIC;
 527         r.hdr.ctdb_version = CTDB_PROTOCOL;
 528         r.hdr.generation = 1;
 529         r.hdr.operation  = CTDB_REQ_MESSAGE;
 530         r.hdr.destnode   = dst_vnn;
 531         r.hdr.srcnode    = conn->our_vnn;
 532         r.hdr.reqid      = 0;
 533         r.srvid          = dst_srvid;
 534         r.datalen        = buflen;
 535
 536         DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
 537         ctdb_packet_dump(&r.hdr);
 538
 539         iov2[0].iov_base = &r;
 540         iov2[0].iov_len = offsetof(struct ctdb_req_message_old, data);
 541         memcpy(&iov2[1], iov, iovlen * sizeof(struct iovec));
 542
 543         nwritten = write_data_iov(conn->fd, iov2, iovlen+1);
 544         if (nwritten == -1) {
 545                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
 546                 cluster_fatal("cluster dispatch daemon msg write error\n");
 547         }
 548
 549         return 0;
 550 }
 551
 552 /*
 553  * send/recv a generic ctdb control message
 554  */
 555 static int ctdbd_control(struct ctdbd_connection *conn,
 556                          uint32_t vnn, uint32_t opcode,
 557                          uint64_t srvid, uint32_t flags,
 558                          TDB_DATA data,
 559                          TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
 560                          int32_t *cstatus)
 561 {
 562         struct ctdb_req_control_old req;
 563         struct ctdb_req_header *hdr;
 564         struct ctdb_reply_control_old *reply = NULL;
 565         struct iovec iov[2];
 566         ssize_t nwritten;
 567         int ret;
 568
 569         ZERO_STRUCT(req);
 570         req.hdr.length = offsetof(struct ctdb_req_control_old, data) + data.dsize;
 571         req.hdr.ctdb_magic   = CTDB_MAGIC;
 572         req.hdr.ctdb_version = CTDB_PROTOCOL;
 573         req.hdr.operation    = CTDB_REQ_CONTROL;
 574         req.hdr.reqid        = ctdbd_next_reqid(conn);
 575         req.hdr.destnode     = vnn;
 576         req.opcode           = opcode;
 577         req.srvid            = srvid;
 578         req.datalen          = data.dsize;
 579         req.flags            = flags;
 580
 581         DBG_DEBUG("Sending ctdb packet reqid=%"PRIu32", vnn=%"PRIu32", "
 582                   "opcode=%"PRIu32", srvid=%"PRIu64"\n", req.hdr.reqid,
 583                   req.hdr.destnode, req.opcode, req.srvid);
 584         ctdb_packet_dump(&req.hdr);
 585
 586         iov[0].iov_base = &req;
 587         iov[0].iov_len = offsetof(struct ctdb_req_control_old, data);
 588         iov[1].iov_base = data.dptr;
 589         iov[1].iov_len = data.dsize;
 590
 591         nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
 592         if (nwritten == -1) {
 593                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
 594                 cluster_fatal("cluster dispatch daemon msg write error\n");
 595         }
 596
 597         if (flags & CTDB_CTRL_FLAG_NOREPLY) {
 598                 if (cstatus) {
 599                         *cstatus = 0;
 600                 }
 601                 return 0;
 602         }
 603
 604         ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
 605         if (ret != 0) {
 606                 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
 607                 return ret;
 608         }
 609
 610         if (hdr->operation != CTDB_REPLY_CONTROL) {
 611                 DEBUG(0, ("received invalid reply\n"));
 612                 TALLOC_FREE(hdr);
 613                 return EIO;
 614         }
 615         reply = (struct ctdb_reply_control_old *)hdr;
 616
 617         if (outdata) {
 618                 if (!(outdata->dptr = (uint8_t *)talloc_memdup(
 619                               mem_ctx, reply->data, reply->datalen))) {
 620                         TALLOC_FREE(reply);
 621                         return ENOMEM;
 622                 }
 623                 outdata->dsize = reply->datalen;
 624         }
 625         if (cstatus) {
 626                 (*cstatus) = reply->status;
 627         }
 628
 629         TALLOC_FREE(reply);
 630         return ret;
 631 }
 632
 633 /*
 634  * see if a remote process exists
 635  */
 636 bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32_t vnn, pid_t pid)
 637 {
 638         int32_t cstatus = 0;
 639         int ret;
 640
 641         ret = ctdbd_control(conn, vnn, CTDB_CONTROL_PROCESS_EXISTS, 0, 0,
 642                             (TDB_DATA) { .dptr = (uint8_t *)&pid,
 643                                          .dsize = sizeof(pid) },
 644                             NULL, NULL, &cstatus);
 645         if (ret != 0) {
 646                 return false;
 647         }
 648         return (cstatus == 0);
 649 }
 650
 651 /*
 652  * Get a db path
 653  */
 654 char *ctdbd_dbpath(struct ctdbd_connection *conn,
 655                    TALLOC_CTX *mem_ctx, uint32_t db_id)
 656 {
 657         int ret;
 658         TDB_DATA data;
 659         TDB_DATA rdata = {0};
 660         int32_t cstatus = 0;
 661
 662         data.dptr = (uint8_t*)&db_id;
 663         data.dsize = sizeof(db_id);
 664
 665         ret = ctdbd_control_local(conn, CTDB_CONTROL_GETDBPATH, 0, 0, data,
 666                                   mem_ctx, &rdata, &cstatus);
 667         if ((ret != 0) || cstatus != 0) {
 668                 DEBUG(0, (__location__ " ctdb_control for getdbpath failed: %s\n",
 669                           strerror(ret)));
 670                 return NULL;
 671         }
 672
 673         return (char *)rdata.dptr;
 674 }
 675
 676 /*
 677  * attach to a ctdb database
 678  */
 679 int ctdbd_db_attach(struct ctdbd_connection *conn,
 680                     const char *name, uint32_t *db_id, int tdb_flags)
 681 {
 682         int ret;
 683         TDB_DATA data;
 684         int32_t cstatus;
 685         bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
 686
 687         data = string_term_tdb_data(name);
 688
 689         ret = ctdbd_control_local(conn,
 690                                   persistent
 691                                   ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
 692                                   : CTDB_CONTROL_DB_ATTACH,
 693                                   tdb_flags, 0, data, NULL, &data, &cstatus);
 694         if (ret != 0) {
 695                 DEBUG(0, (__location__ " ctdb_control for db_attach "
 696                           "failed: %s\n", strerror(ret)));
 697                 return ret;
 698         }
 699
 700         if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
 701                 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
 702                 return EIO;
 703         }
 704
 705         *db_id = *(uint32_t *)data.dptr;
 706         talloc_free(data.dptr);
 707
 708         if (!(tdb_flags & TDB_SEQNUM)) {
 709                 return 0;
 710         }
 711
 712         data.dptr = (uint8_t *)db_id;
 713         data.dsize = sizeof(*db_id);
 714
 715         ret = ctdbd_control_local(conn, CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
 716                                   NULL, NULL, &cstatus);
 717         if ((ret != 0) || cstatus != 0) {
 718                 DEBUG(0, (__location__ " ctdb_control for enable seqnum "
 719                           "failed: %s\n", strerror(ret)));
 720                 return (ret == 0) ? EIO : ret;
 721         }
 722
 723         return 0;
 724 }
 725
 726 /*
 727  * force the migration of a record to this node
 728  */
 729 int ctdbd_migrate(struct ctdbd_connection *conn, uint32_t db_id, TDB_DATA key)
 730 {
 731         struct ctdb_req_call_old req;
 732         struct ctdb_req_header *hdr;
 733         struct iovec iov[2];
 734         ssize_t nwritten;
 735         int ret;
 736
 737         ZERO_STRUCT(req);
 738
 739         req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
 740         req.hdr.ctdb_magic   = CTDB_MAGIC;
 741         req.hdr.ctdb_version = CTDB_PROTOCOL;
 742         req.hdr.operation    = CTDB_REQ_CALL;
 743         req.hdr.reqid        = ctdbd_next_reqid(conn);
 744         req.flags            = CTDB_IMMEDIATE_MIGRATION;
 745         req.callid           = CTDB_NULL_FUNC;
 746         req.db_id            = db_id;
 747         req.keylen           = key.dsize;
 748
 749         DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
 750         ctdb_packet_dump(&req.hdr);
 751
 752         iov[0].iov_base = &req;
 753         iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
 754         iov[1].iov_base = key.dptr;
 755         iov[1].iov_len = key.dsize;
 756
 757         nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
 758         if (nwritten == -1) {
 759                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
 760                 cluster_fatal("cluster dispatch daemon msg write error\n");
 761         }
 762
 763         ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
 764         if (ret != 0) {
 765                 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
 766                 goto fail;
 767         }
 768
 769         if (hdr->operation != CTDB_REPLY_CALL) {
 770                 DEBUG(0, ("received invalid reply\n"));
 771                 goto fail;
 772         }
 773
 774  fail:
 775
 776         TALLOC_FREE(hdr);
 777         return ret;
 778 }
 779
 780 /*
 781  * Fetch a record and parse it
 782  */
 783 int ctdbd_parse(struct ctdbd_connection *conn, uint32_t db_id,
 784                 TDB_DATA key, bool local_copy,
 785                 void (*parser)(TDB_DATA key, TDB_DATA data,
 786                                void *private_data),
 787                 void *private_data)
 788 {
 789         struct ctdb_req_call_old req;
 790         struct ctdb_req_header *hdr = NULL;
 791         struct ctdb_reply_call_old *reply;
 792         struct iovec iov[2];
 793         ssize_t nwritten;
 794         uint32_t flags;
 795         int ret;
 796
 797         flags = local_copy ? CTDB_WANT_READONLY : 0;
 798
 799         ZERO_STRUCT(req);
 800
 801         req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
 802         req.hdr.ctdb_magic   = CTDB_MAGIC;
 803         req.hdr.ctdb_version = CTDB_PROTOCOL;
 804         req.hdr.operation    = CTDB_REQ_CALL;
 805         req.hdr.reqid        = ctdbd_next_reqid(conn);
 806         req.flags            = flags;
 807         req.callid           = CTDB_FETCH_FUNC;
 808         req.db_id            = db_id;
 809         req.keylen           = key.dsize;
 810
 811         iov[0].iov_base = &req;
 812         iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
 813         iov[1].iov_base = key.dptr;
 814         iov[1].iov_len = key.dsize;
 815
 816         nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
 817         if (nwritten == -1) {
 818                 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
 819                 cluster_fatal("cluster dispatch daemon msg write error\n");
 820         }
 821
 822         ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
 823         if (ret != 0) {
 824                 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
 825                 goto fail;
 826         }
 827
 828         if ((hdr == NULL) || (hdr->operation != CTDB_REPLY_CALL)) {
 829                 DEBUG(0, ("received invalid reply\n"));
 830                 ret = EIO;
 831                 goto fail;
 832         }
 833         reply = (struct ctdb_reply_call_old *)hdr;
 834
 835         if (reply->datalen == 0) {
 836                 /*
 837                  * Treat an empty record as non-existing
 838                  */
 839                 ret = ENOENT;
 840                 goto fail;
 841         }
 842
 843         parser(key, make_tdb_data(&reply->data[0], reply->datalen),
 844                private_data);
 845
 846         ret = 0;
 847  fail:
 848         TALLOC_FREE(hdr);
 849         return ret;
 850 }
 851
 852 /*
 853   Traverse a ctdb database. "conn" must be an otherwise unused
 854   ctdb_connection where no other messages but the traverse ones are
 855   expected.
 856 */
 857
 858 int ctdbd_traverse(struct ctdbd_connection *conn, uint32_t db_id,
 859                         void (*fn)(TDB_DATA key, TDB_DATA data,
 860                                    void *private_data),
 861                         void *private_data)
 862 {
 863         int ret;
 864         TDB_DATA key, data;
 865         struct ctdb_traverse_start t;
 866         int32_t cstatus;
 867
 868         t.db_id = db_id;
 869         t.srvid = conn->rand_srvid;
 870         t.reqid = ctdbd_next_reqid(conn);
 871
 872         data.dptr = (uint8_t *)&t;
 873         data.dsize = sizeof(t);
 874
 875         ret = ctdbd_control_local(conn, CTDB_CONTROL_TRAVERSE_START,
 876                                   conn->rand_srvid,
 877                                   0, data, NULL, NULL, &cstatus);
 878
 879         if ((ret != 0) || (cstatus != 0)) {
 880                 DEBUG(0,("ctdbd_control failed: %s, %d\n", strerror(ret),
 881                          cstatus));
 882
 883                 if (ret == 0) {
 884                         /*
 885                          * We need a mapping here
 886                          */
 887                         ret = EIO;
 888                 }
 889                 return ret;
 890         }
 891
 892         while (true) {
 893                 struct ctdb_req_header *hdr = NULL;
 894                 struct ctdb_req_message_old *m;
 895                 struct ctdb_rec_data_old *d;
 896
 897                 ret = ctdb_read_packet(conn->fd, conn->timeout, conn, &hdr);
 898                 if (ret != 0) {
 899                         DEBUG(0, ("ctdb_read_packet failed: %s\n",
 900                                   strerror(ret)));
 901                         cluster_fatal("ctdbd died\n");
 902                 }
 903
 904                 if (hdr->operation != CTDB_REQ_MESSAGE) {
 905                         DEBUG(0, ("Got operation %u, expected a message\n",
 906                                   (unsigned)hdr->operation));
 907                         return EIO;
 908                 }
 909
 910                 m = (struct ctdb_req_message_old *)hdr;
 911                 d = (struct ctdb_rec_data_old *)&m->data[0];
 912                 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
 913                         DEBUG(0, ("Got invalid traverse data of length %d\n",
 914                                   (int)m->datalen));
 915                         return EIO;
 916                 }
 917
 918                 key.dsize = d->keylen;
 919                 key.dptr  = &d->data[0];
 920                 data.dsize = d->datalen;
 921                 data.dptr = &d->data[d->keylen];
 922
 923                 if (key.dsize == 0 && data.dsize == 0) {
 924                         /* end of traverse */
 925                         return 0;
 926                 }
 927
 928                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
 929                         DEBUG(0, ("Got invalid ltdb header length %d\n",
 930                                   (int)data.dsize));
 931                         return EIO;
 932                 }
 933                 data.dsize -= sizeof(struct ctdb_ltdb_header);
 934                 data.dptr += sizeof(struct ctdb_ltdb_header);
 935
 936                 if (fn != NULL) {
 937                         fn(key, data, private_data);
 938                 }
 939         }
 940         return 0;
 941 }
 942
 943 /*
 944    This is used to canonicalize a ctdb_sock_addr structure.
 945 */
 946 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
 947                                       struct sockaddr_storage *out)
 948 {
 949         memcpy(out, in, sizeof (*out));
 950
 951 #ifdef HAVE_IPV6
 952         if (in->ss_family == AF_INET6) {
 953                 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
 954                 const struct sockaddr_in6 *in6 =
 955                         (const struct sockaddr_in6 *)in;
 956                 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
 957                 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
 958                         memset(out, 0, sizeof(*out));
 959 #ifdef HAVE_SOCK_SIN_LEN
 960                         out4->sin_len = sizeof(*out);
 961 #endif
 962                         out4->sin_family = AF_INET;
 963                         out4->sin_port   = in6->sin6_port;
 964                         memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr[12], 4);
 965                 }
 966         }
 967 #endif
 968 }
 969
 970 /*
 971  * Register us as a server for a particular tcp connection
 972  */
 973
 974 int ctdbd_register_ips(struct ctdbd_connection *conn,
 975                        const struct sockaddr_storage *_server,
 976                        const struct sockaddr_storage *_client,
 977                        int (*cb)(uint32_t src_vnn, uint32_t dst_vnn,
 978                                  uint64_t dst_srvid,
 979                                  const uint8_t *msg, size_t msglen,
 980                                  void *private_data),
 981                        void *private_data)
 982 {
 983         struct ctdb_connection p;
 984         TDB_DATA data = { .dptr = (uint8_t *)&p, .dsize = sizeof(p) };
 985         int ret;
 986         struct sockaddr_storage client;
 987         struct sockaddr_storage server;
 988
 989         /*
 990          * Only one connection so far
 991          */
 992
 993         smbd_ctdb_canonicalize_ip(_client, &client);
 994         smbd_ctdb_canonicalize_ip(_server, &server);
 995
 996         switch (client.ss_family) {
 997         case AF_INET:
 998                 memcpy(&p.dst.ip, &server, sizeof(p.dst.ip));
 999                 memcpy(&p.src.ip, &client, sizeof(p.src.ip));
1000                 break;
1001         case AF_INET6:
1002                 memcpy(&p.dst.ip6, &server, sizeof(p.dst.ip6));
1003                 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1004                 break;
1005         default:
1006                 return EIO;
1007         }
1008
1009         /*
1010          * We want to be told about IP releases
1011          */
1012
1013         ret = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP,
1014                                   cb, private_data);
1015         if (ret != 0) {
1016                 return ret;
1017         }
1018
1019         /*
1020          * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1021          * can send an extra ack to trigger a reset for our client, so it
1022          * immediately reconnects
1023          */
1024         ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
1025                             CTDB_CONTROL_TCP_CLIENT, 0,
1026                             CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL,
1027                             NULL);
1028         if (ret != 0) {
1029                 return ret;
1030         }
1031         return 0;
1032 }
1033
1034 /*
1035   call a control on the local node
1036  */
1037 int ctdbd_control_local(struct ctdbd_connection *conn, uint32_t opcode,
1038                         uint64_t srvid, uint32_t flags, TDB_DATA data,
1039                         TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1040                         int32_t *cstatus)
1041 {
1042         return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data,
1043                              mem_ctx, outdata, cstatus);
1044 }
1045
1046 int ctdb_watch_us(struct ctdbd_connection *conn)
1047 {
1048         struct ctdb_notify_data_old reg_data;
1049         size_t struct_len;
1050         int ret;
1051         int32_t cstatus;
1052
1053         reg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1054         reg_data.len = 1;
1055         reg_data.notify_data[0] = 0;
1056
1057         struct_len = offsetof(struct ctdb_notify_data_old,
1058                               notify_data) + reg_data.len;
1059
1060         ret = ctdbd_control_local(
1061                 conn, CTDB_CONTROL_REGISTER_NOTIFY, conn->rand_srvid, 0,
1062                 make_tdb_data((uint8_t *)&reg_data, struct_len),
1063                 NULL, NULL, &cstatus);
1064         if (ret != 0) {
1065                 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1066                           strerror(ret)));
1067         }
1068         return ret;
1069 }
1070
1071 int ctdb_unwatch(struct ctdbd_connection *conn)
1072 {
1073         uint64_t srvid = CTDB_SRVID_SAMBA_NOTIFY;
1074         int ret;
1075         int32_t cstatus;
1076
1077         ret = ctdbd_control_local(
1078                 conn, CTDB_CONTROL_DEREGISTER_NOTIFY, conn->rand_srvid, 0,
1079                 make_tdb_data((uint8_t *)&srvid, sizeof(srvid)),
1080                 NULL, NULL, &cstatus);
1081         if (ret != 0) {
1082                 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1083                           strerror(ret)));
1084         }
1085         return ret;
1086 }
1087
1088 int ctdbd_probe(const char *sockname, int timeout)
1089 {
1090         /*
1091          * Do a very early check if ctdbd is around to avoid an abort and core
1092          * later
1093          */
1094         struct ctdbd_connection *conn = NULL;
1095         int ret;
1096
1097         ret = ctdbd_init_connection(talloc_tos(), sockname, timeout,
1098                                     &conn);
1099
1100         /*
1101          * We only care if we can connect.
1102          */
1103         TALLOC_FREE(conn);
1104
1105         return ret;
1106 }