ctdb/tests/src/fake_ctdbd.c

   1 /*
   2    Fake CTDB server for testing
   3
   4    Copyright (C) Amitay Isaacs  2016
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "replace.h"
  21 #include "system/network.h"
  22 #include "system/time.h"
  23
  24 #include <popt.h>
  25 #include <talloc.h>
  26 #include <tevent.h>
  27 #include <tdb.h>
  28
  29 #include "lib/util/dlinklist.h"
  30 #include "lib/util/tevent_unix.h"
  31 #include "lib/util/debug.h"
  32 #include "lib/util/samba_util.h"
  33 #include "lib/async_req/async_sock.h"
  34
  35 #include "protocol/protocol.h"
  36 #include "protocol/protocol_api.h"
  37 #include "protocol/protocol_util.h"
  38
  39 #include "common/comm.h"
  40 #include "common/logging.h"
  41 #include "common/tunable.h"
  42 #include "common/srvid.h"
  43
  44 #include "ipalloc_read_known_ips.h"
  45
  46
  47 #define CTDB_PORT 4379
  48
  49 /* A fake flag that is only supported by some functions */
  50 #define NODE_FLAGS_FAKE_TIMEOUT 0x80000000
  51
  52 struct node {
  53         ctdb_sock_addr addr;
  54         uint32_t pnn;
  55         uint32_t flags;
  56         uint32_t capabilities;
  57         bool recovery_disabled;
  58         void *recovery_substate;
  59 };
  60
  61 struct node_map {
  62         uint32_t num_nodes;
  63         struct node *node;
  64         uint32_t pnn;
  65         uint32_t recmaster;
  66 };
  67
  68 struct interface {
  69         const char *name;
  70         bool link_up;
  71         uint32_t references;
  72 };
  73
  74 struct interface_map {
  75         int num;
  76         struct interface *iface;
  77 };
  78
  79 struct vnn_map {
  80         uint32_t recmode;
  81         uint32_t generation;
  82         uint32_t size;
  83         uint32_t *map;
  84 };
  85
  86 struct database {
  87         struct database *prev, *next;
  88         const char *name;
  89         uint32_t id;
  90         uint8_t flags;
  91         uint64_t seq_num;
  92 };
  93
  94 struct database_map {
  95         struct database *db;
  96 };
  97
  98 struct fake_control_failure {
  99         struct fake_control_failure  *prev, *next;
 100         enum ctdb_controls opcode;
 101         uint32_t pnn;
 102         const char *error;
 103         const char *comment;
 104 };
 105
 106 struct ctdb_client {
 107         struct ctdb_client *prev, *next;
 108         struct ctdbd_context *ctdb;
 109         pid_t pid;
 110         void *state;
 111 };
 112
 113 struct ctdbd_context {
 114         struct node_map *node_map;
 115         struct interface_map *iface_map;
 116         struct vnn_map *vnn_map;
 117         struct database_map *db_map;
 118         struct srvid_context *srv;
 119         int num_clients;
 120         struct timeval start_time;
 121         struct timeval recovery_start_time;
 122         struct timeval recovery_end_time;
 123         bool takeover_disabled;
 124         int log_level;
 125         enum ctdb_runstate runstate;
 126         struct ctdb_tunable_list tun_list;
 127         char *reclock;
 128         struct ctdb_public_ip_list *known_ips;
 129         struct fake_control_failure *control_failures;
 130         struct ctdb_client *client_list;
 131 };
 132
 133 /*
 134  * Parse routines
 135  */
 136
 137 static struct node_map *nodemap_init(TALLOC_CTX *mem_ctx)
 138 {
 139         struct node_map *node_map;
 140
 141         node_map = talloc_zero(mem_ctx, struct node_map);
 142         if (node_map == NULL) {
 143                 return NULL;
 144         }
 145
 146         node_map->pnn = CTDB_UNKNOWN_PNN;
 147         node_map->recmaster = CTDB_UNKNOWN_PNN;
 148
 149         return node_map;
 150 }
 151
 152 /* Read a nodemap from stdin.  Each line looks like:
 153  *  <PNN> <FLAGS> [RECMASTER] [CURRENT] [CAPABILITIES]
 154  * EOF or a blank line terminates input.
 155  *
 156  * By default, capablities for each node are
 157  * CTDB_CAP_RECMASTER|CTDB_CAP_LMASTER.  These 2
 158  * capabilities can be faked off by adding, for example,
 159  * -CTDB_CAP_RECMASTER.
 160  */
 161
 162 static bool nodemap_parse(struct node_map *node_map)
 163 {
 164         char line[1024];
 165
 166         while ((fgets(line, sizeof(line), stdin) != NULL)) {
 167                 uint32_t pnn, flags, capabilities;
 168                 char *tok, *t;
 169                 char *ip;
 170                 ctdb_sock_addr saddr;
 171                 struct node *node;
 172                 int ret;
 173
 174                 if (line[0] == '\n') {
 175                         break;
 176                 }
 177
 178                 /* Get rid of pesky newline */
 179                 if ((t = strchr(line, '\n')) != NULL) {
 180                         *t = '\0';
 181                 }
 182
 183                 /* Get PNN */
 184                 tok = strtok(line, " \t");
 185                 if (tok == NULL) {
 186                         fprintf(stderr, "bad line (%s) - missing PNN\n", line);
 187                         continue;
 188                 }
 189                 pnn = (uint32_t)strtoul(tok, NULL, 0);
 190
 191                 /* Get IP */
 192                 tok = strtok(NULL, " \t");
 193                 if (tok == NULL) {
 194                         fprintf(stderr, "bad line (%s) - missing IP\n", line);
 195                         continue;
 196                 }
 197                 ret = ctdb_sock_addr_from_string(tok, &saddr, false);
 198                 if (ret != 0) {
 199                         fprintf(stderr, "bad line (%s) - invalid IP\n", line);
 200                         continue;
 201                 }
 202                 ctdb_sock_addr_set_port(&saddr, CTDB_PORT);
 203                 ip = talloc_strdup(node_map, tok);
 204                 if (ip == NULL) {
 205                         goto fail;
 206                 }
 207
 208                 /* Get flags */
 209                 tok = strtok(NULL, " \t");
 210                 if (tok == NULL) {
 211                         fprintf(stderr, "bad line (%s) - missing flags\n",
 212                                 line);
 213                         continue;
 214                 }
 215                 flags = (uint32_t)strtoul(tok, NULL, 0);
 216                 /* Handle deleted nodes */
 217                 if (flags & NODE_FLAGS_DELETED) {
 218                         talloc_free(ip);
 219                         ip = talloc_strdup(node_map, "0.0.0.0");
 220                         if (ip == NULL) {
 221                                 goto fail;
 222                         }
 223                 }
 224                 capabilities = CTDB_CAP_RECMASTER|CTDB_CAP_LMASTER;
 225
 226                 tok = strtok(NULL, " \t");
 227                 while (tok != NULL) {
 228                         if (strcmp(tok, "CURRENT") == 0) {
 229                                 node_map->pnn = pnn;
 230                         } else if (strcmp(tok, "RECMASTER") == 0) {
 231                                 node_map->recmaster = pnn;
 232                         } else if (strcmp(tok, "-CTDB_CAP_RECMASTER") == 0) {
 233                                 capabilities &= ~CTDB_CAP_RECMASTER;
 234                         } else if (strcmp(tok, "-CTDB_CAP_LMASTER") == 0) {
 235                                 capabilities &= ~CTDB_CAP_LMASTER;
 236                         } else if (strcmp(tok, "TIMEOUT") == 0) {
 237                                 /* This can be done with just a flag
 238                                  * value but it is probably clearer
 239                                  * and less error-prone to fake this
 240                                  * with an explicit token */
 241                                 flags |= NODE_FLAGS_FAKE_TIMEOUT;
 242                         }
 243                         tok = strtok(NULL, " \t");
 244                 }
 245
 246                 node_map->node = talloc_realloc(node_map, node_map->node,
 247                                                 struct node,
 248                                                 node_map->num_nodes + 1);
 249                 if (node_map->node == NULL) {
 250                         goto fail;
 251                 }
 252                 node = &node_map->node[node_map->num_nodes];
 253
 254                 ret = ctdb_sock_addr_from_string(ip, &node->addr, false);
 255                 if (ret != 0) {
 256                         fprintf(stderr, "bad line (%s) - invalid IP\n", line);
 257                         continue;
 258                 }
 259                 ctdb_sock_addr_set_port(&node->addr, CTDB_PORT);
 260                 node->pnn = pnn;
 261                 node->flags = flags;
 262                 node->capabilities = capabilities;
 263                 node->recovery_disabled = false;
 264                 node->recovery_substate = NULL;
 265
 266                 node_map->num_nodes += 1;
 267         }
 268
 269         DEBUG(DEBUG_INFO, ("Parsing nodemap done\n"));
 270         return true;
 271
 272 fail:
 273         DEBUG(DEBUG_INFO, ("Parsing nodemap failed\n"));
 274         return false;
 275
 276 }
 277
 278 /* Append a node to a node map with given address and flags */
 279 static bool node_map_add(struct ctdb_node_map *nodemap,
 280                          const char *nstr, uint32_t flags)
 281 {
 282         ctdb_sock_addr addr;
 283         uint32_t num;
 284         struct ctdb_node_and_flags *n;
 285         int ret;
 286
 287         ret = ctdb_sock_addr_from_string(nstr, &addr, false);
 288         if (ret != 0) {
 289                 fprintf(stderr, "Invalid IP address %s\n", nstr);
 290                 return false;
 291         }
 292         ctdb_sock_addr_set_port(&addr, CTDB_PORT);
 293
 294         num = nodemap->num;
 295         nodemap->node = talloc_realloc(nodemap, nodemap->node,
 296                                        struct ctdb_node_and_flags, num+1);
 297         if (nodemap->node == NULL) {
 298                 return false;
 299         }
 300
 301         n = &nodemap->node[num];
 302         n->addr = addr;
 303         n->pnn = num;
 304         n->flags = flags;
 305
 306         nodemap->num = num+1;
 307         return true;
 308 }
 309
 310 /* Read a nodes file into a node map */
 311 static struct ctdb_node_map *ctdb_read_nodes_file(TALLOC_CTX *mem_ctx,
 312                                                   const char *nlist)
 313 {
 314         char **lines;
 315         int nlines;
 316         int i;
 317         struct ctdb_node_map *nodemap;
 318
 319         nodemap = talloc_zero(mem_ctx, struct ctdb_node_map);
 320         if (nodemap == NULL) {
 321                 return NULL;
 322         }
 323
 324         lines = file_lines_load(nlist, &nlines, 0, mem_ctx);
 325         if (lines == NULL) {
 326                 return NULL;
 327         }
 328
 329         while (nlines > 0 && strcmp(lines[nlines-1], "") == 0) {
 330                 nlines--;
 331         }
 332
 333         for (i=0; i<nlines; i++) {
 334                 char *node;
 335                 uint32_t flags;
 336                 size_t len;
 337
 338                 node = lines[i];
 339                 /* strip leading spaces */
 340                 while((*node == ' ') || (*node == '\t')) {
 341                         node++;
 342                 }
 343
 344                 len = strlen(node);
 345
 346                 /* strip trailing spaces */
 347                 while ((len > 1) &&
 348                        ((node[len-1] == ' ') || (node[len-1] == '\t')))
 349                 {
 350                         node[len-1] = '\0';
 351                         len--;
 352                 }
 353
 354                 if (len == 0) {
 355                         continue;
 356                 }
 357                 if (*node == '#') {
 358                         /* A "deleted" node is a node that is
 359                            commented out in the nodes file.  This is
 360                            used instead of removing a line, which
 361                            would cause subsequent nodes to change
 362                            their PNN. */
 363                         flags = NODE_FLAGS_DELETED;
 364                         node = discard_const("0.0.0.0");
 365                 } else {
 366                         flags = 0;
 367                 }
 368                 if (! node_map_add(nodemap, node, flags)) {
 369                         talloc_free(lines);
 370                         TALLOC_FREE(nodemap);
 371                         return NULL;
 372                 }
 373         }
 374
 375         talloc_free(lines);
 376         return nodemap;
 377 }
 378
 379 static struct ctdb_node_map *read_nodes_file(TALLOC_CTX *mem_ctx,
 380                                              uint32_t pnn)
 381 {
 382         struct ctdb_node_map *nodemap;
 383         char nodes_list[PATH_MAX];
 384         const char *ctdb_base;
 385         int num;
 386
 387         ctdb_base = getenv("CTDB_BASE");
 388         if (ctdb_base == NULL) {
 389                 D_ERR("CTDB_BASE is not set\n");
 390                 return NULL;
 391         }
 392
 393         /* read optional node-specific nodes file */
 394         num = snprintf(nodes_list, sizeof(nodes_list),
 395                        "%s/nodes.%d", ctdb_base, pnn);
 396         if (num == sizeof(nodes_list)) {
 397                 D_ERR("nodes file path too long\n");
 398                 return NULL;
 399         }
 400         nodemap = ctdb_read_nodes_file(mem_ctx, nodes_list);
 401         if (nodemap != NULL) {
 402                 /* Fake a load failure for an empty nodemap */
 403                 if (nodemap->num == 0) {
 404                         talloc_free(nodemap);
 405
 406                         D_ERR("Failed to read nodes file \"%s\"\n", nodes_list);
 407                         return NULL;
 408                 }
 409
 410                 return nodemap;
 411         }
 412
 413         /* read normal nodes file */
 414         num = snprintf(nodes_list, sizeof(nodes_list), "%s/nodes", ctdb_base);
 415         if (num == sizeof(nodes_list)) {
 416                 D_ERR("nodes file path too long\n");
 417                 return NULL;
 418         }
 419         nodemap = ctdb_read_nodes_file(mem_ctx, nodes_list);
 420         if (nodemap != NULL) {
 421                 return nodemap;
 422         }
 423
 424         DBG_ERR("Failed to read nodes file \"%s\"\n", nodes_list);
 425         return NULL;
 426 }
 427
 428 static struct interface_map *interfaces_init(TALLOC_CTX *mem_ctx)
 429 {
 430         struct interface_map *iface_map;
 431
 432         iface_map = talloc_zero(mem_ctx, struct interface_map);
 433         if (iface_map == NULL) {
 434                 return NULL;
 435         }
 436
 437         return iface_map;
 438 }
 439
 440 /* Read interfaces information.  Same format as "ctdb ifaces -Y"
 441  * output:
 442  *   :Name:LinkStatus:References:
 443  *   :eth2:1:4294967294
 444  *   :eth1:1:4294967292
 445  */
 446
 447 static bool interfaces_parse(struct interface_map *iface_map)
 448 {
 449         char line[1024];
 450
 451         while ((fgets(line, sizeof(line), stdin) != NULL)) {
 452                 uint16_t link_state;
 453                 uint32_t references;
 454                 char *tok, *t, *name;
 455                 struct interface *iface;
 456
 457                 if (line[0] == '\n') {
 458                         break;
 459                 }
 460
 461                 /* Get rid of pesky newline */
 462                 if ((t = strchr(line, '\n')) != NULL) {
 463                         *t = '\0';
 464                 }
 465
 466                 if (strcmp(line, ":Name:LinkStatus:References:") == 0) {
 467                         continue;
 468                 }
 469
 470                 /* Leading colon... */
 471                 // tok = strtok(line, ":");
 472
 473                 /* name */
 474                 tok = strtok(line, ":");
 475                 if (tok == NULL) {
 476                         fprintf(stderr, "bad line (%s) - missing name\n", line);
 477                         continue;
 478                 }
 479                 name = tok;
 480
 481                 /* link_state */
 482                 tok = strtok(NULL, ":");
 483                 if (tok == NULL) {
 484                         fprintf(stderr, "bad line (%s) - missing link state\n",
 485                                 line);
 486                         continue;
 487                 }
 488                 link_state = (uint16_t)strtoul(tok, NULL, 0);
 489
 490                 /* references... */
 491                 tok = strtok(NULL, ":");
 492                 if (tok == NULL) {
 493                         fprintf(stderr, "bad line (%s) - missing references\n",
 494                                 line);
 495                         continue;
 496                 }
 497                 references = (uint32_t)strtoul(tok, NULL, 0);
 498
 499                 iface_map->iface = talloc_realloc(iface_map, iface_map->iface,
 500                                                   struct interface,
 501                                                   iface_map->num + 1);
 502                 if (iface_map->iface == NULL) {
 503                         goto fail;
 504                 }
 505
 506                 iface = &iface_map->iface[iface_map->num];
 507
 508                 iface->name = talloc_strdup(iface_map, name);
 509                 if (iface->name == NULL) {
 510                         goto fail;
 511                 }
 512                 iface->link_up = link_state;
 513                 iface->references = references;
 514
 515                 iface_map->num += 1;
 516         }
 517
 518         DEBUG(DEBUG_INFO, ("Parsing interfaces done\n"));
 519         return true;
 520
 521 fail:
 522         fprintf(stderr, "Parsing interfaces failed\n");
 523         return false;
 524 }
 525
 526 static struct vnn_map *vnnmap_init(TALLOC_CTX *mem_ctx)
 527 {
 528         struct vnn_map *vnn_map;
 529
 530         vnn_map = talloc_zero(mem_ctx, struct vnn_map);
 531         if (vnn_map == NULL) {
 532                 fprintf(stderr, "Memory error\n");
 533                 return NULL;
 534         }
 535         vnn_map->recmode = CTDB_RECOVERY_ACTIVE;
 536         vnn_map->generation = INVALID_GENERATION;
 537
 538         return vnn_map;
 539 }
 540
 541 /* Read vnn map.
 542  * output:
 543  *   <GENERATION>
 544  *   <LMASTER0>
 545  *   <LMASTER1>
 546  *   ...
 547  */
 548
 549 static bool vnnmap_parse(struct vnn_map *vnn_map)
 550 {
 551         char line[1024];
 552
 553         while (fgets(line, sizeof(line), stdin) != NULL) {
 554                 uint32_t n;
 555                 char *t;
 556
 557                 if (line[0] == '\n') {
 558                         break;
 559                 }
 560
 561                 /* Get rid of pesky newline */
 562                 if ((t = strchr(line, '\n')) != NULL) {
 563                         *t = '\0';
 564                 }
 565
 566                 n = (uint32_t) strtol(line, NULL, 0);
 567
 568                 /* generation */
 569                 if (vnn_map->generation == INVALID_GENERATION) {
 570                         vnn_map->generation = n;
 571                         continue;
 572                 }
 573
 574                 vnn_map->map = talloc_realloc(vnn_map, vnn_map->map, uint32_t,
 575                                               vnn_map->size + 1);
 576                 if (vnn_map->map == NULL) {
 577                         fprintf(stderr, "Memory error\n");
 578                         goto fail;
 579                 }
 580
 581                 vnn_map->map[vnn_map->size] = n;
 582                 vnn_map->size += 1;
 583         }
 584
 585         DEBUG(DEBUG_INFO, ("Parsing vnnmap done\n"));
 586         return true;
 587
 588 fail:
 589         fprintf(stderr, "Parsing vnnmap failed\n");
 590         return false;
 591 }
 592
 593 static bool reclock_parse(struct ctdbd_context *ctdb)
 594 {
 595         char line[1024];
 596         char *t;
 597
 598         if (fgets(line, sizeof(line), stdin) == NULL) {
 599                 goto fail;
 600         }
 601
 602         if (line[0] == '\n') {
 603                 /* Recovery lock remains unset */
 604                 goto ok;
 605         }
 606
 607         /* Get rid of pesky newline */
 608         if ((t = strchr(line, '\n')) != NULL) {
 609                 *t = '\0';
 610         }
 611
 612         ctdb->reclock = talloc_strdup(ctdb, line);
 613         if (ctdb->reclock == NULL) {
 614                 goto fail;
 615         }
 616 ok:
 617         /* Swallow possible blank line following section.  Picky
 618          * compiler settings don't allow the return value to be
 619          * ignored, so make the compiler happy.
 620          */
 621         if (fgets(line, sizeof(line), stdin) == NULL) {
 622                 ;
 623         }
 624         DEBUG(DEBUG_INFO, ("Parsing reclock done\n"));
 625         return true;
 626
 627 fail:
 628         fprintf(stderr, "Parsing reclock failed\n");
 629         return false;
 630 }
 631
 632 static struct database_map *dbmap_init(TALLOC_CTX *mem_ctx)
 633 {
 634         struct database_map *db_map;
 635
 636         db_map = talloc_zero(mem_ctx, struct database_map);
 637         if (db_map == NULL) {
 638                 return NULL;
 639         }
 640
 641         return db_map;
 642 }
 643
 644 /* Read a database map from stdin.  Each line looks like:
 645  *  <ID> <NAME> [FLAGS] [SEQ_NUM]
 646  * EOF or a blank line terminates input.
 647  *
 648  * By default, flags and seq_num are 0
 649  */
 650
 651 static bool dbmap_parse(struct database_map *db_map)
 652 {
 653         char line[1024];
 654
 655         while ((fgets(line, sizeof(line), stdin) != NULL)) {
 656                 uint32_t id;
 657                 uint8_t flags = 0;
 658                 uint32_t seq_num = 0;
 659                 char *tok, *t;
 660                 char *name;
 661                 struct database *db;
 662
 663                 if (line[0] == '\n') {
 664                         break;
 665                 }
 666
 667                 /* Get rid of pesky newline */
 668                 if ((t = strchr(line, '\n')) != NULL) {
 669                         *t = '\0';
 670                 }
 671
 672                 /* Get ID */
 673                 tok = strtok(line, " \t");
 674                 if (tok == NULL) {
 675                         fprintf(stderr, "bad line (%s) - missing ID\n", line);
 676                         continue;
 677                 }
 678                 id = (uint32_t)strtoul(tok, NULL, 0);
 679
 680                 /* Get NAME */
 681                 tok = strtok(NULL, " \t");
 682                 if (tok == NULL) {
 683                         fprintf(stderr, "bad line (%s) - missing NAME\n", line);
 684                         continue;
 685                 }
 686                 name = talloc_strdup(db_map, tok);
 687                 if (name == NULL) {
 688                         goto fail;
 689                 }
 690
 691                 /* Get flags */
 692                 tok = strtok(NULL, " \t");
 693                 while (tok != NULL) {
 694                         if (strcmp(tok, "PERSISTENT") == 0) {
 695                                 flags |= CTDB_DB_FLAGS_PERSISTENT;
 696                         } else if (strcmp(tok, "STICKY") == 0) {
 697                                 flags |= CTDB_DB_FLAGS_STICKY;
 698                         } else if (strcmp(tok, "READONLY") == 0) {
 699                                 flags |= CTDB_DB_FLAGS_READONLY;
 700                         } else if (strcmp(tok, "REPLICATED") == 0) {
 701                                 flags |= CTDB_DB_FLAGS_REPLICATED;
 702                         } else if (tok[0] >= '0'&& tok[0] <= '9') {
 703                                 uint8_t nv = CTDB_DB_FLAGS_PERSISTENT |
 704                                              CTDB_DB_FLAGS_REPLICATED;
 705
 706                                 if ((flags & nv) == 0) {
 707                                         fprintf(stderr,
 708                                                 "seq_num for volatile db\n");
 709                                         goto fail;
 710                                 }
 711                                 seq_num = (uint64_t)strtoull(tok, NULL, 0);
 712                         }
 713
 714                         tok = strtok(NULL, " \t");
 715                 }
 716
 717                 db = talloc_zero(db_map, struct database);
 718                 if (db == NULL) {
 719                         goto fail;
 720                 }
 721
 722                 db->id = id;
 723                 db->name = talloc_steal(db, name);
 724                 db->flags = flags;
 725                 db->seq_num = seq_num;
 726
 727                 DLIST_ADD_END(db_map->db, db);
 728         }
 729
 730         DEBUG(DEBUG_INFO, ("Parsing dbmap done\n"));
 731         return true;
 732
 733 fail:
 734         DEBUG(DEBUG_INFO, ("Parsing dbmap failed\n"));
 735         return false;
 736
 737 }
 738
 739 static struct database *database_find(struct database_map *db_map,
 740                                       uint32_t db_id)
 741 {
 742         struct database *db;
 743
 744         for (db = db_map->db; db != NULL; db = db->next) {
 745                 if (db->id == db_id) {
 746                         return db;
 747                 }
 748         }
 749
 750         return NULL;
 751 }
 752
 753 static int database_count(struct database_map *db_map)
 754 {
 755         struct database *db;
 756         int count = 0;
 757
 758         for (db = db_map->db; db != NULL; db = db->next) {
 759                 count += 1;
 760         }
 761
 762         return count;
 763 }
 764
 765 static bool public_ips_parse(struct ctdbd_context *ctdb,
 766                              uint32_t numnodes)
 767 {
 768         bool status;
 769
 770         if (numnodes == 0) {
 771                 D_ERR("Must initialise nodemap before public IPs\n");
 772                 return false;
 773         }
 774
 775         ctdb->known_ips = ipalloc_read_known_ips(ctdb, numnodes, false);
 776
 777         status = (ctdb->known_ips != NULL);
 778
 779         if (status) {
 780                 D_INFO("Parsing public IPs done\n");
 781         } else {
 782                 D_INFO("Parsing public IPs failed\n");
 783         }
 784
 785         return status;
 786 }
 787
 788 /* Read information about controls to fail.  Format is:
 789  *   <opcode> <pnn> {ERROR|TIMEOUT} <comment>
 790  */
 791 static bool control_failures_parse(struct ctdbd_context *ctdb)
 792 {
 793         char line[1024];
 794
 795         while ((fgets(line, sizeof(line), stdin) != NULL)) {
 796                 char *tok, *t;
 797                 enum ctdb_controls opcode;
 798                 uint32_t pnn;
 799                 const char *error;
 800                 const char *comment;
 801                 struct fake_control_failure *failure = NULL;
 802
 803                 if (line[0] == '\n') {
 804                         break;
 805                 }
 806
 807                 /* Get rid of pesky newline */
 808                 if ((t = strchr(line, '\n')) != NULL) {
 809                         *t = '\0';
 810                 }
 811
 812                 /* Get opcode */
 813                 tok = strtok(line, " \t");
 814                 if (tok == NULL) {
 815                         D_ERR("bad line (%s) - missing opcode\n", line);
 816                         continue;
 817                 }
 818                 opcode = (enum ctdb_controls)strtoul(tok, NULL, 0);
 819
 820                 /* Get PNN */
 821                 tok = strtok(NULL, " \t");
 822                 if (tok == NULL) {
 823                         D_ERR("bad line (%s) - missing PNN\n", line);
 824                         continue;
 825                 }
 826                 pnn = (uint32_t)strtoul(tok, NULL, 0);
 827
 828                 /* Get error */
 829                 tok = strtok(NULL, " \t");
 830                 if (tok == NULL) {
 831                         D_ERR("bad line (%s) - missing errno\n", line);
 832                         continue;
 833                 }
 834                 error = talloc_strdup(ctdb, tok);
 835                 if (error == NULL) {
 836                         goto fail;
 837                 }
 838                 if (strcmp(error, "ERROR") != 0 &&
 839                     strcmp(error, "TIMEOUT") != 0) {
 840                         D_ERR("bad line (%s) "
 841                               "- error must be \"ERROR\" or \"TIMEOUT\"\n",
 842                               line);
 843                         goto fail;
 844                 }
 845
 846                 /* Get comment */
 847                 tok = strtok(NULL, "\n"); /* rest of line */
 848                 if (tok == NULL) {
 849                         D_ERR("bad line (%s) - missing comment\n", line);
 850                         continue;
 851                 }
 852                 comment = talloc_strdup(ctdb, tok);
 853                 if (comment == NULL) {
 854                         goto fail;
 855                 }
 856
 857                 failure = talloc_zero(ctdb, struct fake_control_failure);
 858                 if (failure == NULL) {
 859                         goto fail;
 860                 }
 861
 862                 failure->opcode = opcode;
 863                 failure->pnn = pnn;
 864                 failure->error = error;
 865                 failure->comment = comment;
 866
 867                 DLIST_ADD(ctdb->control_failures, failure);
 868         }
 869
 870         D_INFO("Parsing fake control failures done\n");
 871         return true;
 872
 873 fail:
 874         D_INFO("Parsing fake control failures failed\n");
 875         return false;
 876 }
 877
 878 /*
 879  * Manage clients
 880  */
 881
 882 static int ctdb_client_destructor(struct ctdb_client *client)
 883 {
 884         DLIST_REMOVE(client->ctdb->client_list, client);
 885         return 0;
 886 }
 887
 888 static int client_add(struct ctdbd_context *ctdb, pid_t client_pid,
 889                       void *client_state)
 890 {
 891         struct ctdb_client *client;
 892
 893         client = talloc_zero(client_state, struct ctdb_client);
 894         if (client == NULL) {
 895                 return ENOMEM;
 896         }
 897
 898         client->ctdb = ctdb;
 899         client->pid = client_pid;
 900         client->state = client_state;
 901
 902         DLIST_ADD(ctdb->client_list, client);
 903         talloc_set_destructor(client, ctdb_client_destructor);
 904         return 0;
 905 }
 906
 907 static void *client_find(struct ctdbd_context *ctdb, pid_t client_pid)
 908 {
 909         struct ctdb_client *client;
 910
 911         for (client=ctdb->client_list; client != NULL; client=client->next) {
 912                 if (client->pid == client_pid) {
 913                         return client->state;
 914                 }
 915         }
 916
 917         return NULL;
 918 }
 919
 920 /*
 921  * CTDB context setup
 922  */
 923
 924 static uint32_t new_generation(uint32_t old_generation)
 925 {
 926         uint32_t generation;
 927
 928         while (1) {
 929                 generation = random();
 930                 if (generation != INVALID_GENERATION &&
 931                     generation != old_generation) {
 932                         break;
 933                 }
 934         }
 935
 936         return generation;
 937 }
 938
 939 static struct ctdbd_context *ctdbd_setup(TALLOC_CTX *mem_ctx)
 940 {
 941         struct ctdbd_context *ctdb;
 942         char line[1024];
 943         bool status;
 944         int ret;
 945
 946         ctdb = talloc_zero(mem_ctx, struct ctdbd_context);
 947         if (ctdb == NULL) {
 948                 return NULL;
 949         }
 950
 951         ctdb->node_map = nodemap_init(ctdb);
 952         if (ctdb->node_map == NULL) {
 953                 goto fail;
 954         }
 955
 956         ctdb->iface_map = interfaces_init(ctdb);
 957         if (ctdb->iface_map == NULL) {
 958                 goto fail;
 959         }
 960
 961         ctdb->vnn_map = vnnmap_init(ctdb);
 962         if (ctdb->vnn_map == NULL) {
 963                 goto fail;
 964         }
 965
 966         ctdb->db_map = dbmap_init(ctdb);
 967         if (ctdb->db_map == NULL) {
 968                 goto fail;
 969         }
 970
 971         ret = srvid_init(ctdb, &ctdb->srv);
 972         if (ret != 0) {
 973                 goto fail;
 974         }
 975
 976         while (fgets(line, sizeof(line), stdin) != NULL) {
 977                 char *t;
 978
 979                 if ((t = strchr(line, '\n')) != NULL) {
 980                         *t = '\0';
 981                 }
 982
 983                 if (strcmp(line, "NODEMAP") == 0) {
 984                         status = nodemap_parse(ctdb->node_map);
 985                 } else if (strcmp(line, "IFACES") == 0) {
 986                         status = interfaces_parse(ctdb->iface_map);
 987                 } else if (strcmp(line, "VNNMAP") == 0) {
 988                         status = vnnmap_parse(ctdb->vnn_map);
 989                 } else if (strcmp(line, "DBMAP") == 0) {
 990                         status = dbmap_parse(ctdb->db_map);
 991                 } else if (strcmp(line, "PUBLICIPS") == 0) {
 992                         status = public_ips_parse(ctdb,
 993                                                   ctdb->node_map->num_nodes);
 994                 } else if (strcmp(line, "RECLOCK") == 0) {
 995                         status = reclock_parse(ctdb);
 996                 } else if (strcmp(line, "CONTROLFAILS") == 0) {
 997                         status = control_failures_parse(ctdb);
 998                 } else {
 999                         fprintf(stderr, "Unknown line %s\n", line);
1000                         status = false;
1001                 }
1002
1003                 if (! status) {
1004                         goto fail;
1005                 }
1006         }
1007
1008         ctdb->start_time = tevent_timeval_current();
1009         ctdb->recovery_start_time = tevent_timeval_current();
1010         ctdb->vnn_map->recmode = CTDB_RECOVERY_NORMAL;
1011         if (ctdb->vnn_map->generation == INVALID_GENERATION) {
1012                 ctdb->vnn_map->generation =
1013                         new_generation(ctdb->vnn_map->generation);
1014         }
1015         ctdb->recovery_end_time = tevent_timeval_current();
1016
1017         ctdb->log_level = DEBUG_ERR;
1018         ctdb->runstate = CTDB_RUNSTATE_RUNNING;
1019
1020         ctdb_tunable_set_defaults(&ctdb->tun_list);
1021
1022         return ctdb;
1023
1024 fail:
1025         TALLOC_FREE(ctdb);
1026         return NULL;
1027 }
1028
1029 static bool ctdbd_verify(struct ctdbd_context *ctdb)
1030 {
1031         struct node *node;
1032         int i;
1033
1034         if (ctdb->node_map->num_nodes == 0) {
1035                 return true;
1036         }
1037
1038         /* Make sure all the nodes are in order */
1039         for (i=0; i<ctdb->node_map->num_nodes; i++) {
1040                 node = &ctdb->node_map->node[i];
1041                 if (node->pnn != i) {
1042                         fprintf(stderr, "Expected node %u, found %u\n",
1043                                 i, node->pnn);
1044                         return false;
1045                 }
1046         }
1047
1048         node = &ctdb->node_map->node[ctdb->node_map->pnn];
1049         if (node->flags & NODE_FLAGS_DISCONNECTED) {
1050                 DEBUG(DEBUG_INFO, ("Node disconnected, exiting\n"));
1051                 exit(0);
1052         }
1053
1054         return true;
1055 }
1056
1057 /*
1058  * Doing a recovery
1059  */
1060
1061 struct recover_state {
1062         struct tevent_context *ev;
1063         struct ctdbd_context *ctdb;
1064 };
1065
1066 static int recover_check(struct tevent_req *req);
1067 static void recover_wait_done(struct tevent_req *subreq);
1068 static void recover_done(struct tevent_req *subreq);
1069
1070 static struct tevent_req *recover_send(TALLOC_CTX *mem_ctx,
1071                                        struct tevent_context *ev,
1072                                        struct ctdbd_context *ctdb)
1073 {
1074         struct tevent_req *req;
1075         struct recover_state *state;
1076         int ret;
1077
1078         req = tevent_req_create(mem_ctx, &state, struct recover_state);
1079         if (req == NULL) {
1080                 return NULL;
1081         }
1082
1083         state->ev = ev;
1084         state->ctdb = ctdb;
1085
1086         ret = recover_check(req);
1087         if (ret != 0) {
1088                 tevent_req_error(req, ret);
1089                 return tevent_req_post(req, ev);
1090         }
1091
1092         return req;
1093 }
1094
1095 static int recover_check(struct tevent_req *req)
1096 {
1097         struct recover_state *state = tevent_req_data(
1098                 req, struct recover_state);
1099         struct ctdbd_context *ctdb = state->ctdb;
1100         struct tevent_req *subreq;
1101         bool recovery_disabled;
1102         int i;
1103
1104         recovery_disabled = false;
1105         for (i=0; i<ctdb->node_map->num_nodes; i++) {
1106                 if (ctdb->node_map->node[i].recovery_disabled) {
1107                         recovery_disabled = true;
1108                         break;
1109                 }
1110         }
1111
1112         subreq = tevent_wakeup_send(state, state->ev,
1113                                     tevent_timeval_current_ofs(1, 0));
1114         if (subreq == NULL) {
1115                 return ENOMEM;
1116         }
1117
1118         if (recovery_disabled) {
1119                 tevent_req_set_callback(subreq, recover_wait_done, req);
1120         } else {
1121                 ctdb->recovery_start_time = tevent_timeval_current();
1122                 tevent_req_set_callback(subreq, recover_done, req);
1123         }
1124
1125         return 0;
1126 }
1127
1128 static void recover_wait_done(struct tevent_req *subreq)
1129 {
1130         struct tevent_req *req = tevent_req_callback_data(
1131                 subreq, struct tevent_req);
1132         int ret;
1133         bool status;
1134
1135         status = tevent_wakeup_recv(subreq);
1136         TALLOC_FREE(subreq);
1137         if (! status) {
1138                 tevent_req_error(req, EIO);
1139                 return;
1140         }
1141
1142         ret = recover_check(req);
1143         if (ret != 0) {
1144                 tevent_req_error(req, ret);
1145         }
1146 }
1147
1148 static void recover_done(struct tevent_req *subreq)
1149 {
1150         struct tevent_req *req = tevent_req_callback_data(
1151                 subreq, struct tevent_req);
1152         struct recover_state *state = tevent_req_data(
1153                 req, struct recover_state);
1154         struct ctdbd_context *ctdb = state->ctdb;
1155         bool status;
1156
1157         status = tevent_wakeup_recv(subreq);
1158         TALLOC_FREE(subreq);
1159         if (! status) {
1160                 tevent_req_error(req, EIO);
1161                 return;
1162         }
1163
1164         ctdb->vnn_map->recmode = CTDB_RECOVERY_NORMAL;
1165         ctdb->recovery_end_time = tevent_timeval_current();
1166         ctdb->vnn_map->generation = new_generation(ctdb->vnn_map->generation);
1167
1168         tevent_req_done(req);
1169 }
1170
1171 static bool recover_recv(struct tevent_req *req, int *perr)
1172 {
1173         int err;
1174
1175         if (tevent_req_is_unix_error(req, &err)) {
1176                 if (perr != NULL) {
1177                         *perr = err;
1178                 }
1179                 return false;
1180         }
1181
1182         return true;
1183 }
1184
1185 /*
1186  * Routines for ctdb_req_header
1187  */
1188
1189 static void header_fix_pnn(struct ctdb_req_header *header,
1190                            struct ctdbd_context *ctdb)
1191 {
1192         if (header->srcnode == CTDB_CURRENT_NODE) {
1193                 header->srcnode = ctdb->node_map->pnn;
1194         }
1195
1196         if (header->destnode == CTDB_CURRENT_NODE) {
1197                 header->destnode = ctdb->node_map->pnn;
1198         }
1199 }
1200
1201 static struct ctdb_req_header header_reply_control(
1202                                         struct ctdb_req_header *header,
1203                                         struct ctdbd_context *ctdb)
1204 {
1205         struct ctdb_req_header reply_header;
1206
1207         reply_header = (struct ctdb_req_header) {
1208                 .ctdb_magic = CTDB_MAGIC,
1209                 .ctdb_version = CTDB_PROTOCOL,
1210                 .generation = ctdb->vnn_map->generation,
1211                 .operation = CTDB_REPLY_CONTROL,
1212                 .destnode = header->srcnode,
1213                 .srcnode = header->destnode,
1214                 .reqid = header->reqid,
1215         };
1216
1217         return reply_header;
1218 }
1219
1220 static struct ctdb_req_header header_reply_message(
1221                                         struct ctdb_req_header *header,
1222                                         struct ctdbd_context *ctdb)
1223 {
1224         struct ctdb_req_header reply_header;
1225
1226         reply_header = (struct ctdb_req_header) {
1227                 .ctdb_magic = CTDB_MAGIC,
1228                 .ctdb_version = CTDB_PROTOCOL,
1229                 .generation = ctdb->vnn_map->generation,
1230                 .operation = CTDB_REQ_MESSAGE,
1231                 .destnode = header->srcnode,
1232                 .srcnode = header->destnode,
1233                 .reqid = 0,
1234         };
1235
1236         return reply_header;
1237 }
1238
1239 /*
1240  * Client state
1241  */
1242
1243 struct client_state {
1244         struct tevent_context *ev;
1245         int fd;
1246         struct ctdbd_context *ctdb;
1247         int pnn;
1248         pid_t pid;
1249         struct comm_context *comm;
1250         struct srvid_register_state *rstate;
1251         int status;
1252 };
1253
1254 /*
1255  * Send replies to controls and messages
1256  */
1257
1258 static void client_reply_done(struct tevent_req *subreq);
1259
1260 static void client_send_message(struct tevent_req *req,
1261                                 struct ctdb_req_header *header,
1262                                 struct ctdb_req_message_data *message)
1263 {
1264         struct client_state *state = tevent_req_data(
1265                 req, struct client_state);
1266         struct ctdbd_context *ctdb = state->ctdb;
1267         struct tevent_req *subreq;
1268         struct ctdb_req_header reply_header;
1269         uint8_t *buf;
1270         size_t datalen, buflen;
1271         int ret;
1272
1273         reply_header = header_reply_message(header, ctdb);
1274
1275         datalen = ctdb_req_message_data_len(&reply_header, message);
1276         ret = ctdb_allocate_pkt(state, datalen, &buf, &buflen);
1277         if (ret != 0) {
1278                 tevent_req_error(req, ret);
1279                 return;
1280         }
1281
1282         ret = ctdb_req_message_data_push(&reply_header, message,
1283                                          buf, &buflen);
1284         if (ret != 0) {
1285                 tevent_req_error(req, ret);
1286                 return;
1287         }
1288
1289         DEBUG(DEBUG_INFO, ("message srvid = 0x%"PRIx64"\n", message->srvid));
1290
1291         subreq = comm_write_send(state, state->ev, state->comm, buf, buflen);
1292         if (tevent_req_nomem(subreq, req)) {
1293                 return;
1294         }
1295         tevent_req_set_callback(subreq, client_reply_done, req);
1296
1297         talloc_steal(subreq, buf);
1298 }
1299
1300 static void client_send_control(struct tevent_req *req,
1301                                 struct ctdb_req_header *header,
1302                                 struct ctdb_reply_control *reply)
1303 {
1304         struct client_state *state = tevent_req_data(
1305                 req, struct client_state);
1306         struct ctdbd_context *ctdb = state->ctdb;
1307         struct tevent_req *subreq;
1308         struct ctdb_req_header reply_header;
1309         uint8_t *buf;
1310         size_t datalen, buflen;
1311         int ret;
1312
1313         reply_header = header_reply_control(header, ctdb);
1314
1315         datalen = ctdb_reply_control_len(&reply_header, reply);
1316         ret = ctdb_allocate_pkt(state, datalen, &buf, &buflen);
1317         if (ret != 0) {
1318                 tevent_req_error(req, ret);
1319                 return;
1320         }
1321
1322         ret = ctdb_reply_control_push(&reply_header, reply, buf, &buflen);
1323         if (ret != 0) {
1324                 tevent_req_error(req, ret);
1325                 return;
1326         }
1327
1328         DEBUG(DEBUG_INFO, ("reply opcode = %u\n", reply->rdata.opcode));
1329
1330         subreq = comm_write_send(state, state->ev, state->comm, buf, buflen);
1331         if (tevent_req_nomem(subreq, req)) {
1332                 return;
1333         }
1334         tevent_req_set_callback(subreq, client_reply_done, req);
1335
1336         talloc_steal(subreq, buf);
1337 }
1338
1339 static void client_reply_done(struct tevent_req *subreq)
1340 {
1341         struct tevent_req *req = tevent_req_callback_data(
1342                 subreq, struct tevent_req);
1343         int ret;
1344         bool status;
1345
1346         status = comm_write_recv(subreq, &ret);
1347         TALLOC_FREE(subreq);
1348         if (! status) {
1349                 tevent_req_error(req, ret);
1350         }
1351 }
1352
1353 /*
1354  * Handling protocol - controls
1355  */
1356
1357 static void control_process_exists(TALLOC_CTX *mem_ctx,
1358                                    struct tevent_req *req,
1359                                    struct ctdb_req_header *header,
1360                                    struct ctdb_req_control *request)
1361 {
1362         struct client_state *state = tevent_req_data(
1363                 req, struct client_state);
1364         struct ctdbd_context *ctdb = state->ctdb;
1365         struct client_state *cstate;
1366         struct ctdb_reply_control reply;
1367
1368         reply.rdata.opcode = request->opcode;
1369
1370         cstate = client_find(ctdb, request->rdata.data.pid);
1371         if (cstate == NULL) {
1372                 reply.status = -1;
1373                 reply.errmsg = "No client for PID";
1374         } else {
1375                 reply.status = kill(request->rdata.data.pid, 0);
1376                 reply.errmsg = NULL;
1377         }
1378
1379         client_send_control(req, header, &reply);
1380 }
1381
1382 static void control_ping(TALLOC_CTX *mem_ctx,
1383                          struct tevent_req *req,
1384                          struct ctdb_req_header *header,
1385                          struct ctdb_req_control *request)
1386 {
1387         struct client_state *state = tevent_req_data(
1388                 req, struct client_state);
1389         struct ctdbd_context *ctdb = state->ctdb;
1390         struct ctdb_reply_control reply;
1391
1392         reply.rdata.opcode = request->opcode;
1393         reply.status = ctdb->num_clients;
1394         reply.errmsg = NULL;
1395
1396         client_send_control(req, header, &reply);
1397 }
1398
1399 static void control_getdbpath(TALLOC_CTX *mem_ctx,
1400                               struct tevent_req *req,
1401                               struct ctdb_req_header *header,
1402                               struct ctdb_req_control *request)
1403 {
1404         struct client_state *state = tevent_req_data(
1405                 req, struct client_state);
1406         struct ctdbd_context *ctdb = state->ctdb;
1407         struct ctdb_reply_control reply;
1408         struct database *db;
1409
1410         reply.rdata.opcode = request->opcode;
1411
1412         db = database_find(ctdb->db_map, request->rdata.data.db_id);
1413         if (db == NULL) {
1414                 reply.status = ENOENT;
1415                 reply.errmsg = "Database not found";
1416         } else {
1417                 const char *base;
1418                 if (db->flags & CTDB_DB_FLAGS_PERSISTENT) {
1419                         base = "/var/lib/ctdb/persistent";
1420                 } else {
1421                         base = "/var/run/ctdb/DB_DIR";
1422                 }
1423                 reply.rdata.data.db_path =
1424                         talloc_asprintf(mem_ctx, "%s/%s.%u",
1425                                         base, db->name, header->destnode);
1426                 if (reply.rdata.data.db_path == NULL) {
1427                         reply.status = ENOMEM;
1428                         reply.errmsg = "Memory error";
1429                 } else {
1430                         reply.status = 0;
1431                         reply.errmsg = NULL;
1432                 }
1433         }
1434
1435         client_send_control(req, header, &reply);
1436 }
1437
1438 static void control_getvnnmap(TALLOC_CTX *mem_ctx,
1439                               struct tevent_req *req,
1440                               struct ctdb_req_header *header,
1441                               struct ctdb_req_control *request)
1442 {
1443         struct client_state *state = tevent_req_data(
1444                 req, struct client_state);
1445         struct ctdbd_context *ctdb = state->ctdb;
1446         struct ctdb_reply_control reply;
1447         struct ctdb_vnn_map *vnnmap;
1448
1449         reply.rdata.opcode = request->opcode;
1450
1451         vnnmap = talloc_zero(mem_ctx, struct ctdb_vnn_map);
1452         if (vnnmap == NULL) {
1453                 reply.status = ENOMEM;
1454                 reply.errmsg = "Memory error";
1455         } else {
1456                 vnnmap->generation = ctdb->vnn_map->generation;
1457                 vnnmap->size = ctdb->vnn_map->size;
1458                 vnnmap->map = ctdb->vnn_map->map;
1459
1460                 reply.rdata.data.vnnmap = vnnmap;
1461                 reply.status = 0;
1462                 reply.errmsg = NULL;
1463         }
1464
1465         client_send_control(req, header, &reply);
1466 }
1467
1468 static void control_get_debug(TALLOC_CTX *mem_ctx,
1469                               struct tevent_req *req,
1470                               struct ctdb_req_header *header,
1471                               struct ctdb_req_control *request)
1472 {
1473         struct client_state *state = tevent_req_data(
1474                 req, struct client_state);
1475         struct ctdbd_context *ctdb = state->ctdb;
1476         struct ctdb_reply_control reply;
1477
1478         reply.rdata.opcode = request->opcode;
1479         reply.rdata.data.loglevel = (uint32_t)ctdb->log_level;
1480         reply.status = 0;
1481         reply.errmsg = NULL;
1482
1483         client_send_control(req, header, &reply);
1484 }
1485
1486 static void control_set_debug(TALLOC_CTX *mem_ctx,
1487                               struct tevent_req *req,
1488                               struct ctdb_req_header *header,
1489                               struct ctdb_req_control *request)
1490 {
1491         struct client_state *state = tevent_req_data(
1492                 req, struct client_state);
1493         struct ctdbd_context *ctdb = state->ctdb;
1494         struct ctdb_reply_control reply;
1495
1496         ctdb->log_level = (int)request->rdata.data.loglevel;
1497
1498         reply.rdata.opcode = request->opcode;
1499         reply.status = 0;
1500         reply.errmsg = NULL;
1501
1502         client_send_control(req, header, &reply);
1503 }
1504
1505 static void control_get_dbmap(TALLOC_CTX *mem_ctx,
1506                               struct tevent_req *req,
1507                                struct ctdb_req_header *header,
1508                               struct ctdb_req_control *request)
1509 {
1510         struct client_state *state = tevent_req_data(
1511                 req, struct client_state);
1512         struct ctdbd_context *ctdb = state->ctdb;
1513         struct ctdb_reply_control reply;
1514         struct ctdb_dbid_map *dbmap;
1515         struct database *db;
1516         int i;
1517
1518         reply.rdata.opcode = request->opcode;
1519
1520         dbmap = talloc_zero(mem_ctx, struct ctdb_dbid_map);
1521         if (dbmap == NULL) {
1522                 goto fail;
1523         }
1524
1525         dbmap->num = database_count(ctdb->db_map);
1526         dbmap->dbs = talloc_zero_array(dbmap, struct ctdb_dbid, dbmap->num);
1527         if (dbmap->dbs == NULL) {
1528                 goto fail;
1529         }
1530
1531         db = ctdb->db_map->db;
1532         for (i = 0; i < dbmap->num; i++) {
1533                 dbmap->dbs[i] = (struct ctdb_dbid) {
1534                         .db_id = db->id,
1535                         .flags = db->flags,
1536                 };
1537
1538                 db = db->next;
1539         }
1540
1541         reply.rdata.data.dbmap = dbmap;
1542         reply.status = 0;
1543         reply.errmsg = NULL;
1544         client_send_control(req, header, &reply);
1545         return;
1546
1547 fail:
1548         reply.status = -1;
1549         reply.errmsg = "Memory error";
1550         client_send_control(req, header, &reply);
1551 }
1552
1553 static void control_get_recmode(TALLOC_CTX *mem_ctx,
1554                                 struct tevent_req *req,
1555                                 struct ctdb_req_header *header,
1556                                 struct ctdb_req_control *request)
1557 {
1558         struct client_state *state = tevent_req_data(
1559                 req, struct client_state);
1560         struct ctdbd_context *ctdb = state->ctdb;
1561         struct ctdb_reply_control reply;
1562
1563         reply.rdata.opcode = request->opcode;
1564         reply.status = ctdb->vnn_map->recmode;
1565         reply.errmsg = NULL;
1566
1567         client_send_control(req, header, &reply);
1568 }
1569
1570 struct set_recmode_state {
1571         struct tevent_req *req;
1572         struct ctdbd_context *ctdb;
1573         struct ctdb_req_header header;
1574         struct ctdb_reply_control reply;
1575 };
1576
1577 static void set_recmode_callback(struct tevent_req *subreq)
1578 {
1579         struct set_recmode_state *substate = tevent_req_callback_data(
1580                 subreq, struct set_recmode_state);
1581         bool status;
1582         int ret;
1583
1584         status = recover_recv(subreq, &ret);
1585         TALLOC_FREE(subreq);
1586         if (! status) {
1587                 substate->reply.status = ret;
1588                 substate->reply.errmsg = "recovery failed";
1589         } else {
1590                 substate->reply.status = 0;
1591                 substate->reply.errmsg = NULL;
1592         }
1593
1594         client_send_control(substate->req, &substate->header, &substate->reply);
1595         talloc_free(substate);
1596 }
1597
1598 static void control_set_recmode(TALLOC_CTX *mem_ctx,
1599                                 struct tevent_req *req,
1600                                 struct ctdb_req_header *header,
1601                                 struct ctdb_req_control *request)
1602 {
1603         struct client_state *state = tevent_req_data(
1604                 req, struct client_state);
1605         struct tevent_req *subreq;
1606         struct ctdbd_context *ctdb = state->ctdb;
1607         struct set_recmode_state *substate;
1608         struct ctdb_reply_control reply;
1609
1610         reply.rdata.opcode = request->opcode;
1611
1612         if (request->rdata.data.recmode == CTDB_RECOVERY_NORMAL) {
1613                 reply.status = -1;
1614                 reply.errmsg = "Client cannot set recmode to NORMAL";
1615                 goto fail;
1616         }
1617
1618         substate = talloc_zero(ctdb, struct set_recmode_state);
1619         if (substate == NULL) {
1620                 reply.status = -1;
1621                 reply.errmsg = "Memory error";
1622                 goto fail;
1623         }
1624
1625         substate->req = req;
1626         substate->ctdb = ctdb;
1627         substate->header = *header;
1628         substate->reply.rdata.opcode = request->opcode;
1629
1630         subreq = recover_send(substate, state->ev, state->ctdb);
1631         if (subreq == NULL) {
1632                 talloc_free(substate);
1633                 goto fail;
1634         }
1635         tevent_req_set_callback(subreq, set_recmode_callback, substate);
1636
1637         ctdb->vnn_map->recmode = CTDB_RECOVERY_ACTIVE;
1638         return;
1639
1640 fail:
1641         client_send_control(req, header, &reply);
1642
1643 }
1644
1645 static void srvid_handler(uint64_t srvid, TDB_DATA data, void *private_data)
1646 {
1647         printf("Received a message for SRVID 0x%"PRIx64"\n", srvid);
1648 }
1649
1650 static void control_register_srvid(TALLOC_CTX *mem_ctx,
1651                                    struct tevent_req *req,
1652                                    struct ctdb_req_header *header,
1653                                    struct ctdb_req_control *request)
1654 {
1655         struct client_state *state = tevent_req_data(
1656                 req, struct client_state);
1657         struct ctdbd_context *ctdb = state->ctdb;
1658         struct ctdb_reply_control reply;
1659         int ret;
1660
1661         reply.rdata.opcode = request->opcode;
1662
1663         ret = srvid_register(ctdb->srv, state, request->srvid,
1664                              srvid_handler, state);
1665         if (ret != 0) {
1666                 reply.status = -1;
1667                 reply.errmsg = "Memory error";
1668                 goto fail;
1669         }
1670
1671         DEBUG(DEBUG_INFO, ("Register srvid 0x%"PRIx64"\n", request->srvid));
1672
1673         reply.status = 0;
1674         reply.errmsg = NULL;
1675
1676 fail:
1677         client_send_control(req, header, &reply);
1678 }
1679
1680 static void control_deregister_srvid(TALLOC_CTX *mem_ctx,
1681                                      struct tevent_req *req,
1682                                      struct ctdb_req_header *header,
1683                                      struct ctdb_req_control *request)
1684 {
1685         struct client_state *state = tevent_req_data(
1686                 req, struct client_state);
1687         struct ctdbd_context *ctdb = state->ctdb;
1688         struct ctdb_reply_control reply;
1689         int ret;
1690
1691         reply.rdata.opcode = request->opcode;
1692
1693         ret = srvid_deregister(ctdb->srv, request->srvid, state);
1694         if (ret != 0) {
1695                 reply.status = -1;
1696                 reply.errmsg = "srvid not registered";
1697                 goto fail;
1698         }
1699
1700         DEBUG(DEBUG_INFO, ("Deregister srvid 0x%"PRIx64"\n", request->srvid));
1701
1702         reply.status = 0;
1703         reply.errmsg = NULL;
1704
1705 fail:
1706         client_send_control(req, header, &reply);
1707 }
1708
1709 static void control_get_dbname(TALLOC_CTX *mem_ctx,
1710                                struct tevent_req *req,
1711                                struct ctdb_req_header *header,
1712                                struct ctdb_req_control *request)
1713 {
1714         struct client_state *state = tevent_req_data(
1715                 req, struct client_state);
1716         struct ctdbd_context *ctdb = state->ctdb;
1717         struct ctdb_reply_control reply;
1718         struct database *db;
1719
1720         reply.rdata.opcode = request->opcode;
1721
1722         db = database_find(ctdb->db_map, request->rdata.data.db_id);
1723         if (db == NULL) {
1724                 reply.status = ENOENT;
1725                 reply.errmsg = "Database not found";
1726         } else {
1727                 reply.rdata.data.db_name = talloc_strdup(mem_ctx, db->name);
1728                 if (reply.rdata.data.db_name == NULL) {
1729                         reply.status = ENOMEM;
1730                         reply.errmsg = "Memory error";
1731                 } else {
1732                         reply.status = 0;
1733                         reply.errmsg = NULL;
1734                 }
1735         }
1736
1737         client_send_control(req, header, &reply);
1738 }
1739
1740 static void control_get_pid(TALLOC_CTX *mem_ctx,
1741                             struct tevent_req *req,
1742                             struct ctdb_req_header *header,
1743                             struct ctdb_req_control *request)
1744 {
1745         struct ctdb_reply_control reply;
1746
1747         reply.rdata.opcode = request->opcode;
1748         reply.status = getpid();
1749         reply.errmsg = NULL;
1750
1751         client_send_control(req, header, &reply);
1752 }
1753
1754 static void control_get_recmaster(TALLOC_CTX *mem_ctx,
1755                                   struct tevent_req *req,
1756                                   struct ctdb_req_header *header,
1757                                   struct ctdb_req_control *request)
1758 {
1759         struct client_state *state = tevent_req_data(
1760                 req, struct client_state);
1761         struct ctdbd_context *ctdb = state->ctdb;
1762         struct ctdb_reply_control reply;
1763
1764         reply.rdata.opcode = request->opcode;
1765         reply.status = ctdb->node_map->recmaster;
1766         reply.errmsg = NULL;
1767
1768         client_send_control(req, header, &reply);
1769 }
1770
1771 static void control_get_pnn(TALLOC_CTX *mem_ctx,
1772                             struct tevent_req *req,
1773                             struct ctdb_req_header *header,
1774                             struct ctdb_req_control *request)
1775 {
1776         struct ctdb_reply_control reply;
1777
1778         reply.rdata.opcode = request->opcode;
1779         reply.status = header->destnode;
1780         reply.errmsg = NULL;
1781
1782         client_send_control(req, header, &reply);
1783 }
1784
1785 static void control_shutdown(TALLOC_CTX *mem_ctx,
1786                              struct tevent_req *req,
1787                              struct ctdb_req_header *hdr,
1788                              struct ctdb_req_control *request)
1789 {
1790         struct client_state *state = tevent_req_data(
1791                 req, struct client_state);
1792
1793         state->status = 99;
1794 }
1795
1796 static void control_set_tunable(TALLOC_CTX *mem_ctx,
1797                                 struct tevent_req *req,
1798                                 struct ctdb_req_header *header,
1799                                 struct ctdb_req_control *request)
1800 {
1801         struct client_state *state = tevent_req_data(
1802                 req, struct client_state);
1803         struct ctdbd_context *ctdb = state->ctdb;
1804         struct ctdb_reply_control reply;
1805         bool ret, obsolete;
1806
1807         reply.rdata.opcode = request->opcode;
1808         reply.errmsg = NULL;
1809
1810         ret = ctdb_tunable_set_value(&ctdb->tun_list,
1811                                      request->rdata.data.tunable->name,
1812                                      request->rdata.data.tunable->value,
1813                                      &obsolete);
1814         if (! ret) {
1815                 reply.status = -1;
1816         } else if (obsolete) {
1817                 reply.status = 1;
1818         } else {
1819                 reply.status = 0;
1820         }
1821
1822         client_send_control(req, header, &reply);
1823 }
1824
1825 static void control_get_tunable(TALLOC_CTX *mem_ctx,
1826                                 struct tevent_req *req,
1827                                 struct ctdb_req_header *header,
1828                                 struct ctdb_req_control *request)
1829 {
1830         struct client_state *state = tevent_req_data(
1831                 req, struct client_state);
1832         struct ctdbd_context *ctdb = state->ctdb;
1833         struct ctdb_reply_control reply;
1834         uint32_t value;
1835         bool ret;
1836
1837         reply.rdata.opcode = request->opcode;
1838         reply.errmsg = NULL;
1839
1840         ret = ctdb_tunable_get_value(&ctdb->tun_list,
1841                                      request->rdata.data.tun_var, &value);
1842         if (! ret) {
1843                 reply.status = -1;
1844         } else {
1845                 reply.rdata.data.tun_value = value;
1846                 reply.status = 0;
1847         }
1848
1849         client_send_control(req, header, &reply);
1850 }
1851
1852 static void control_list_tunables(TALLOC_CTX *mem_ctx,
1853                                   struct tevent_req *req,
1854                                   struct ctdb_req_header *header,
1855                                   struct ctdb_req_control *request)
1856 {
1857         struct ctdb_reply_control reply;
1858         struct ctdb_var_list *var_list;
1859
1860         reply.rdata.opcode = request->opcode;
1861         reply.errmsg = NULL;
1862
1863         var_list = ctdb_tunable_names(mem_ctx);
1864         if (var_list == NULL) {
1865                 reply.status = -1;
1866         } else {
1867                 reply.rdata.data.tun_var_list = var_list;
1868                 reply.status = 0;
1869         }
1870
1871         client_send_control(req, header, &reply);
1872 }
1873
1874 static void control_modify_flags(TALLOC_CTX *mem_ctx,
1875                                  struct tevent_req *req,
1876                                  struct ctdb_req_header *header,
1877                                  struct ctdb_req_control *request)
1878 {
1879         struct client_state *state = tevent_req_data(
1880                 req, struct client_state);
1881         struct ctdbd_context *ctdb = state->ctdb;
1882         struct ctdb_node_flag_change *change = request->rdata.data.flag_change;
1883         struct ctdb_reply_control reply;
1884         struct node *node;
1885
1886         reply.rdata.opcode = request->opcode;
1887
1888         if ((change->old_flags & ~NODE_FLAGS_PERMANENTLY_DISABLED) ||
1889             (change->new_flags & ~NODE_FLAGS_PERMANENTLY_DISABLED) != 0) {
1890                 DEBUG(DEBUG_INFO,
1891                       ("MODIFY_FLAGS control not for PERMANENTLY_DISABLED\n"));
1892                 reply.status = EINVAL;
1893                 reply.errmsg = "Failed to MODIFY_FLAGS";
1894                 client_send_control(req, header, &reply);
1895                 return;
1896         }
1897
1898         /* There's all sorts of broadcast weirdness here.  Only change
1899          * the specified node, not the destination node of the
1900          * control. */
1901         node = &ctdb->node_map->node[change->pnn];
1902
1903         if ((node->flags &
1904              change->old_flags & NODE_FLAGS_PERMANENTLY_DISABLED) == 0 &&
1905             (change->new_flags & NODE_FLAGS_PERMANENTLY_DISABLED) != 0) {
1906                 DEBUG(DEBUG_INFO,("Disabling node %d\n", header->destnode));
1907                 node->flags |= NODE_FLAGS_PERMANENTLY_DISABLED;
1908                 goto done;
1909         }
1910
1911         if ((node->flags &
1912              change->old_flags & NODE_FLAGS_PERMANENTLY_DISABLED) != 0 &&
1913             (change->new_flags & NODE_FLAGS_PERMANENTLY_DISABLED) == 0) {
1914                 DEBUG(DEBUG_INFO,("Enabling node %d\n", header->destnode));
1915                 node->flags &= ~NODE_FLAGS_PERMANENTLY_DISABLED;
1916                 goto done;
1917         }
1918
1919         DEBUG(DEBUG_INFO, ("Flags unchanged for node %d\n", header->destnode));
1920
1921 done:
1922         reply.status = 0;
1923         reply.errmsg = NULL;
1924         client_send_control(req, header, &reply);
1925 }
1926
1927 static void control_get_all_tunables(TALLOC_CTX *mem_ctx,
1928                                      struct tevent_req *req,
1929                                      struct ctdb_req_header *header,
1930                                      struct ctdb_req_control *request)
1931 {
1932         struct client_state *state = tevent_req_data(
1933                 req, struct client_state);
1934         struct ctdbd_context *ctdb = state->ctdb;
1935         struct ctdb_reply_control reply;
1936
1937         reply.rdata.opcode = request->opcode;
1938         reply.rdata.data.tun_list = &ctdb->tun_list;
1939         reply.status = 0;
1940         reply.errmsg = NULL;
1941
1942         client_send_control(req, header, &reply);
1943 }
1944
1945 static void control_uptime(TALLOC_CTX *mem_ctx,
1946                            struct tevent_req *req,
1947                            struct ctdb_req_header *header,
1948                            struct ctdb_req_control *request)
1949 {
1950         struct client_state *state = tevent_req_data(
1951                 req, struct client_state);
1952         struct ctdbd_context *ctdb = state->ctdb;
1953         struct ctdb_reply_control reply;
1954         struct ctdb_uptime *uptime;;
1955
1956         reply.rdata.opcode = request->opcode;
1957
1958         uptime = talloc_zero(mem_ctx, struct ctdb_uptime);
1959         if (uptime == NULL) {
1960                 goto fail;
1961         }
1962
1963         uptime->current_time = tevent_timeval_current();
1964         uptime->ctdbd_start_time = ctdb->start_time;
1965         uptime->last_recovery_started = ctdb->recovery_start_time;
1966         uptime->last_recovery_finished = ctdb->recovery_end_time;
1967
1968         reply.rdata.data.uptime = uptime;
1969         reply.status = 0;
1970         reply.errmsg = NULL;
1971         client_send_control(req, header, &reply);
1972         return;
1973
1974 fail:
1975         reply.status = -1;
1976         reply.errmsg = "Memory error";
1977         client_send_control(req, header, &reply);
1978 }
1979
1980 static void control_reload_nodes_file(TALLOC_CTX *mem_ctx,
1981                                       struct tevent_req *req,
1982                                       struct ctdb_req_header *header,
1983                                       struct ctdb_req_control *request)
1984 {
1985         struct client_state *state = tevent_req_data(
1986                 req, struct client_state);
1987         struct ctdbd_context *ctdb = state->ctdb;
1988         struct ctdb_reply_control reply;
1989         struct ctdb_node_map *nodemap;
1990         struct node_map *node_map = ctdb->node_map;
1991         int i;
1992
1993         reply.rdata.opcode = request->opcode;
1994
1995         nodemap = read_nodes_file(mem_ctx, header->destnode);
1996         if (nodemap == NULL) {
1997                 goto fail;
1998         }
1999
2000         for (i=0; i<nodemap->num; i++) {
2001                 struct node *node;
2002
2003                 if (i < node_map->num_nodes &&
2004                     ctdb_sock_addr_same(&nodemap->node[i].addr,
2005                                         &node_map->node[i].addr)) {
2006                         continue;
2007                 }
2008
2009                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
2010                         int ret;
2011
2012                         node = &node_map->node[i];
2013
2014                         node->flags |= NODE_FLAGS_DELETED;
2015                         ret = ctdb_sock_addr_from_string("0.0.0.0", &node->addr,
2016                                                          false);
2017                         if (ret != 0) {
2018                                 /* Can't happen, but Coverity... */
2019                                 goto fail;
2020                         }
2021
2022                         continue;
2023                 }
2024
2025                 if (i < node_map->num_nodes &&
2026                     node_map->node[i].flags & NODE_FLAGS_DELETED) {
2027                         node = &node_map->node[i];
2028
2029                         node->flags &= ~NODE_FLAGS_DELETED;
2030                         node->addr = nodemap->node[i].addr;
2031
2032                         continue;
2033                 }
2034
2035                 node_map->node = talloc_realloc(node_map, node_map->node,
2036                                                 struct node,
2037                                                 node_map->num_nodes+1);
2038                 if (node_map->node == NULL) {
2039                         goto fail;
2040                 }
2041                 node = &node_map->node[node_map->num_nodes];
2042
2043                 node->addr = nodemap->node[i].addr;
2044                 node->pnn = nodemap->node[i].pnn;
2045                 node->flags = 0;
2046                 node->capabilities = CTDB_CAP_DEFAULT;
2047                 node->recovery_disabled = false;
2048                 node->recovery_substate = NULL;
2049
2050                 node_map->num_nodes += 1;
2051         }
2052
2053         talloc_free(nodemap);
2054
2055         reply.status = 0;
2056         reply.errmsg = NULL;
2057         client_send_control(req, header, &reply);
2058         return;
2059
2060 fail:
2061         reply.status = -1;
2062         reply.errmsg = "Memory error";
2063         client_send_control(req, header, &reply);
2064 }
2065
2066 static void control_get_capabilities(TALLOC_CTX *mem_ctx,
2067                                      struct tevent_req *req,
2068                                      struct ctdb_req_header *header,
2069                                      struct ctdb_req_control *request)
2070 {
2071         struct client_state *state = tevent_req_data(
2072                 req, struct client_state);
2073         struct ctdbd_context *ctdb = state->ctdb;
2074         struct ctdb_reply_control reply;
2075         struct node *node;
2076         uint32_t caps = 0;
2077
2078         reply.rdata.opcode = request->opcode;
2079
2080         node = &ctdb->node_map->node[header->destnode];
2081         caps = node->capabilities;
2082
2083         if (node->flags & NODE_FLAGS_FAKE_TIMEOUT) {
2084                 /* Don't send reply */
2085                 return;
2086         }
2087
2088         reply.rdata.data.caps = caps;
2089         reply.status = 0;
2090         reply.errmsg = NULL;
2091
2092         client_send_control(req, header, &reply);
2093 }
2094
2095 static void control_release_ip(TALLOC_CTX *mem_ctx,
2096                                struct tevent_req *req,
2097                                struct ctdb_req_header *header,
2098                                struct ctdb_req_control *request)
2099 {
2100         struct client_state *state = tevent_req_data(
2101                 req, struct client_state);
2102         struct ctdbd_context *ctdb = state->ctdb;
2103         struct ctdb_public_ip *ip = request->rdata.data.pubip;
2104         struct ctdb_reply_control reply;
2105         struct ctdb_public_ip_list *ips = NULL;
2106         struct ctdb_public_ip *t = NULL;
2107         int i;
2108
2109         reply.rdata.opcode = request->opcode;
2110
2111         if (ctdb->known_ips == NULL) {
2112                 D_INFO("RELEASE_IP %s - not a public IP\n",
2113                        ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false));
2114                 goto done;
2115         }
2116
2117         ips = &ctdb->known_ips[header->destnode];
2118
2119         t = NULL;
2120         for (i = 0; i < ips->num; i++) {
2121                 if (ctdb_sock_addr_same_ip(&ips->ip[i].addr, &ip->addr)) {
2122                         t = &ips->ip[i];
2123                         break;
2124                 }
2125         }
2126         if (t == NULL) {
2127                 D_INFO("RELEASE_IP %s - not a public IP\n",
2128                        ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false));
2129                 goto done;
2130         }
2131
2132         if (t->pnn != header->destnode) {
2133                 if (header->destnode == ip->pnn) {
2134                         D_ERR("error: RELEASE_IP %s - to TAKE_IP node %d\n",
2135                               ctdb_sock_addr_to_string(mem_ctx,
2136                                                        &ip->addr, false),
2137                               ip->pnn);
2138                         reply.status = -1;
2139                         reply.errmsg = "RELEASE_IP to TAKE_IP node";
2140                         client_send_control(req, header, &reply);
2141                         return;
2142                 }
2143
2144                 D_INFO("RELEASE_IP %s - to node %d - redundant\n",
2145                        ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false),
2146                        ip->pnn);
2147                 t->pnn = ip->pnn;
2148         } else {
2149                 D_NOTICE("RELEASE_IP %s - to node %d\n",
2150                          ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false),
2151                           ip->pnn);
2152                 t->pnn = ip->pnn;
2153         }
2154
2155 done:
2156         reply.status = 0;
2157         reply.errmsg = NULL;
2158         client_send_control(req, header, &reply);
2159 }
2160
2161 static void control_takeover_ip(TALLOC_CTX *mem_ctx,
2162                                 struct tevent_req *req,
2163                                 struct ctdb_req_header *header,
2164                                 struct ctdb_req_control *request)
2165 {
2166         struct client_state *state = tevent_req_data(
2167                 req, struct client_state);
2168         struct ctdbd_context *ctdb = state->ctdb;
2169         struct ctdb_public_ip *ip = request->rdata.data.pubip;
2170         struct ctdb_reply_control reply;
2171         struct ctdb_public_ip_list *ips = NULL;
2172         struct ctdb_public_ip *t = NULL;
2173         int i;
2174
2175         reply.rdata.opcode = request->opcode;
2176
2177         if (ctdb->known_ips == NULL) {
2178                 D_INFO("TAKEOVER_IP %s - not a public IP\n",
2179                        ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false));
2180                 goto done;
2181         }
2182
2183         ips = &ctdb->known_ips[header->destnode];
2184
2185         t = NULL;
2186         for (i = 0; i < ips->num; i++) {
2187                 if (ctdb_sock_addr_same_ip(&ips->ip[i].addr, &ip->addr)) {
2188                         t = &ips->ip[i];
2189                         break;
2190                 }
2191         }
2192         if (t == NULL) {
2193                 D_INFO("TAKEOVER_IP %s - not a public IP\n",
2194                        ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false));
2195                 goto done;
2196         }
2197
2198         if (t->pnn == header->destnode) {
2199                 D_INFO("TAKEOVER_IP %s - redundant\n",
2200                        ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false));
2201         } else {
2202                 D_NOTICE("TAKEOVER_IP %s\n",
2203                          ctdb_sock_addr_to_string(mem_ctx, &ip->addr, false));
2204                 t->pnn = ip->pnn;
2205         }
2206
2207 done:
2208         reply.status = 0;
2209         reply.errmsg = NULL;
2210         client_send_control(req, header, &reply);
2211 }
2212
2213 static void control_get_public_ips(TALLOC_CTX *mem_ctx,
2214                                    struct tevent_req *req,
2215                                    struct ctdb_req_header *header,
2216                                    struct ctdb_req_control *request)
2217 {
2218         struct client_state *state = tevent_req_data(
2219                 req, struct client_state);
2220         struct ctdbd_context *ctdb = state->ctdb;
2221         struct ctdb_reply_control reply;
2222         struct ctdb_public_ip_list *ips = NULL;
2223
2224         reply.rdata.opcode = request->opcode;
2225
2226         if (ctdb->known_ips == NULL) {
2227                 /* No IPs defined so create a dummy empty struct and ship it */
2228                 ips = talloc_zero(mem_ctx, struct ctdb_public_ip_list);;
2229                 if (ips == NULL) {
2230                         reply.status = ENOMEM;
2231                         reply.errmsg = "Memory error";
2232                         goto done;
2233                 }
2234                 goto ok;
2235         }
2236
2237         ips = &ctdb->known_ips[header->destnode];
2238
2239         if (request->flags & CTDB_PUBLIC_IP_FLAGS_ONLY_AVAILABLE) {
2240                 /* If runstate is not RUNNING or a node is then return
2241                  * no available IPs.  Don't worry about interface
2242                  * states here - we're not faking down to that level.
2243                  */
2244                 if (ctdb->runstate != CTDB_RUNSTATE_RUNNING) {
2245                         /* No available IPs: return dummy empty struct */
2246                         ips = talloc_zero(mem_ctx, struct ctdb_public_ip_list);;
2247                         if (ips == NULL) {
2248                                 reply.status = ENOMEM;
2249                                 reply.errmsg = "Memory error";
2250                                 goto done;
2251                         }
2252                 }
2253         }
2254
2255 ok:
2256         reply.rdata.data.pubip_list = ips;
2257         reply.status = 0;
2258         reply.errmsg = NULL;
2259
2260 done:
2261         client_send_control(req, header, &reply);
2262 }
2263
2264 static void control_get_nodemap(TALLOC_CTX *mem_ctx,
2265                                 struct tevent_req *req,
2266                                 struct ctdb_req_header *header,
2267                                 struct ctdb_req_control *request)
2268 {
2269         struct client_state *state = tevent_req_data(
2270                 req, struct client_state);
2271         struct ctdbd_context *ctdb = state->ctdb;
2272         struct ctdb_reply_control reply;
2273         struct ctdb_node_map *nodemap;
2274         struct node *node;
2275         int i;
2276
2277         reply.rdata.opcode = request->opcode;
2278
2279         nodemap = talloc_zero(mem_ctx, struct ctdb_node_map);
2280         if (nodemap == NULL) {
2281                 goto fail;
2282         }
2283
2284         nodemap->num = ctdb->node_map->num_nodes;
2285         nodemap->node = talloc_array(nodemap, struct ctdb_node_and_flags,
2286                                      nodemap->num);
2287         if (nodemap->node == NULL) {
2288                 goto fail;
2289         }
2290
2291         for (i=0; i<nodemap->num; i++) {
2292                 node = &ctdb->node_map->node[i];
2293                 nodemap->node[i] = (struct ctdb_node_and_flags) {
2294                         .pnn = node->pnn,
2295                         .flags = node->flags,
2296                         .addr = node->addr,
2297                 };
2298         }
2299
2300         reply.rdata.data.nodemap = nodemap;
2301         reply.status = 0;
2302         reply.errmsg = NULL;
2303         client_send_control(req, header, &reply);
2304         return;
2305
2306 fail:
2307         reply.status = -1;
2308         reply.errmsg = "Memory error";
2309         client_send_control(req, header, &reply);
2310 }
2311
2312 static void control_get_reclock_file(TALLOC_CTX *mem_ctx,
2313                                      struct tevent_req *req,
2314                                      struct ctdb_req_header *header,
2315                                      struct ctdb_req_control *request)
2316 {
2317         struct client_state *state = tevent_req_data(
2318                 req, struct client_state);
2319         struct ctdbd_context *ctdb = state->ctdb;
2320         struct ctdb_reply_control reply;
2321
2322         reply.rdata.opcode = request->opcode;
2323
2324         if (ctdb->reclock != NULL) {
2325                 reply.rdata.data.reclock_file =
2326                         talloc_strdup(mem_ctx, ctdb->reclock);
2327                 if (reply.rdata.data.reclock_file == NULL) {
2328                         reply.status = ENOMEM;
2329                         reply.errmsg = "Memory error";
2330                         goto done;
2331                 }
2332         } else {
2333                 reply.rdata.data.reclock_file = NULL;
2334         }
2335
2336         reply.status = 0;
2337         reply.errmsg = NULL;
2338
2339 done:
2340         client_send_control(req, header, &reply);
2341 }
2342
2343 static void control_stop_node(TALLOC_CTX *mem_ctx,
2344                               struct tevent_req *req,
2345                               struct ctdb_req_header *header,
2346                               struct ctdb_req_control *request)
2347 {
2348         struct client_state *state = tevent_req_data(
2349                 req, struct client_state);
2350         struct ctdbd_context *ctdb = state->ctdb;
2351         struct ctdb_reply_control reply;
2352
2353         reply.rdata.opcode = request->opcode;
2354
2355         DEBUG(DEBUG_INFO, ("Stopping node\n"));
2356         ctdb->node_map->node[header->destnode].flags |= NODE_FLAGS_STOPPED;
2357
2358         reply.status = 0;
2359         reply.errmsg = NULL;
2360
2361         client_send_control(req, header, &reply);
2362         return;
2363 }
2364
2365 static void control_continue_node(TALLOC_CTX *mem_ctx,
2366                                   struct tevent_req *req,
2367                                   struct ctdb_req_header *header,
2368                                   struct ctdb_req_control *request)
2369 {
2370         struct client_state *state = tevent_req_data(
2371                 req, struct client_state);
2372         struct ctdbd_context *ctdb = state->ctdb;
2373         struct ctdb_reply_control reply;
2374
2375         reply.rdata.opcode = request->opcode;
2376
2377         DEBUG(DEBUG_INFO, ("Continue node\n"));
2378         ctdb->node_map->node[header->destnode].flags &= ~NODE_FLAGS_STOPPED;
2379
2380         reply.status = 0;
2381         reply.errmsg = NULL;
2382
2383         client_send_control(req, header, &reply);
2384         return;
2385 }
2386
2387 static void set_ban_state_callback(struct tevent_req *subreq)
2388 {
2389         struct node *node = tevent_req_callback_data(
2390                 subreq, struct node);
2391         bool status;
2392
2393         status = tevent_wakeup_recv(subreq);
2394         TALLOC_FREE(subreq);
2395         if (! status) {
2396                 DEBUG(DEBUG_INFO, ("tevent_wakeup_recv failed\n"));
2397         }
2398
2399         node->flags &= ~NODE_FLAGS_BANNED;
2400 }
2401
2402 static void control_set_ban_state(TALLOC_CTX *mem_ctx,
2403                                   struct tevent_req *req,
2404                                   struct ctdb_req_header *header,
2405                                   struct ctdb_req_control *request)
2406 {
2407         struct client_state *state = tevent_req_data(
2408                 req, struct client_state);
2409         struct tevent_req *subreq;
2410         struct ctdbd_context *ctdb = state->ctdb;
2411         struct ctdb_ban_state *ban = request->rdata.data.ban_state;
2412         struct ctdb_reply_control reply;
2413         struct node *node;
2414
2415         reply.rdata.opcode = request->opcode;
2416
2417         if (ban->pnn != header->destnode) {
2418                 DEBUG(DEBUG_INFO,
2419                       ("SET_BAN_STATE control for PNN %d rejected\n",
2420                        ban->pnn));
2421                 reply.status = EINVAL;
2422                 goto fail;
2423         }
2424
2425         node = &ctdb->node_map->node[header->destnode];
2426
2427         if (ban->time == 0) {
2428                 DEBUG(DEBUG_INFO,("Unbanning this node\n"));
2429                 node->flags &= ~NODE_FLAGS_BANNED;
2430                 goto done;
2431         }
2432
2433         subreq = tevent_wakeup_send(ctdb->node_map, state->ev,
2434                                     tevent_timeval_current_ofs(
2435                                             ban->time, 0));
2436         if (subreq == NULL) {
2437                 reply.status = ENOMEM;
2438                 goto fail;
2439         }
2440         tevent_req_set_callback(subreq, set_ban_state_callback, node);
2441
2442         DEBUG(DEBUG_INFO, ("Banning this node for %d seconds\n", ban->time));
2443         node->flags |= NODE_FLAGS_BANNED;
2444         ctdb->vnn_map->generation = INVALID_GENERATION;
2445
2446 done:
2447         reply.status = 0;
2448         reply.errmsg = NULL;
2449
2450         client_send_control(req, header, &reply);
2451         return;
2452
2453 fail:
2454         reply.errmsg = "Failed to ban node";
2455 }
2456
2457 static void control_get_db_seqnum(TALLOC_CTX *mem_ctx,
2458                                struct tevent_req *req,
2459                                struct ctdb_req_header *header,
2460                                struct ctdb_req_control *request)
2461 {
2462         struct client_state *state = tevent_req_data(
2463                 req, struct client_state);
2464         struct ctdbd_context *ctdb = state->ctdb;
2465         struct ctdb_reply_control reply;
2466         struct database *db;
2467
2468         reply.rdata.opcode = request->opcode;
2469
2470         db = database_find(ctdb->db_map, request->rdata.data.db_id);
2471         if (db == NULL) {
2472                 reply.status = ENOENT;
2473                 reply.errmsg = "Database not found";
2474         } else {
2475                 reply.rdata.data.seqnum = db->seq_num;
2476                 reply.status = 0;
2477                 reply.errmsg = NULL;
2478         }
2479
2480         client_send_control(req, header, &reply);
2481 }
2482
2483 static void control_db_get_health(TALLOC_CTX *mem_ctx,
2484                                   struct tevent_req *req,
2485                                   struct ctdb_req_header *header,
2486                                   struct ctdb_req_control *request)
2487 {
2488         struct client_state *state = tevent_req_data(
2489                 req, struct client_state);
2490         struct ctdbd_context *ctdb = state->ctdb;
2491         struct ctdb_reply_control reply;
2492         struct database *db;
2493
2494         reply.rdata.opcode = request->opcode;
2495
2496         db = database_find(ctdb->db_map, request->rdata.data.db_id);
2497         if (db == NULL) {
2498                 reply.status = ENOENT;
2499                 reply.errmsg = "Database not found";
2500         } else {
2501                 reply.rdata.data.reason = NULL;
2502                 reply.status = 0;
2503                 reply.errmsg = NULL;
2504         }
2505
2506         client_send_control(req, header, &reply);
2507 }
2508
2509 static struct ctdb_iface_list *get_ctdb_iface_list(TALLOC_CTX *mem_ctx,
2510                                                    struct ctdbd_context *ctdb)
2511 {
2512         struct ctdb_iface_list *iface_list;
2513         struct interface *iface;
2514         int i;
2515
2516         iface_list = talloc_zero(mem_ctx, struct ctdb_iface_list);
2517         if (iface_list == NULL) {
2518                 goto done;
2519         }
2520
2521         iface_list->num = ctdb->iface_map->num;
2522         iface_list->iface = talloc_array(iface_list, struct ctdb_iface,
2523                                          iface_list->num);
2524         if (iface_list->iface == NULL) {
2525                 TALLOC_FREE(iface_list);
2526                 goto done;
2527         }
2528
2529         for (i=0; i<iface_list->num; i++) {
2530                 iface = &ctdb->iface_map->iface[i];
2531                 iface_list->iface[i] = (struct ctdb_iface) {
2532                         .link_state = iface->link_up,
2533                         .references = iface->references,
2534                 };
2535                 strlcpy(iface_list->iface[i].name, iface->name,
2536                         sizeof(iface_list->iface[i].name));
2537         }
2538
2539 done:
2540         return iface_list;
2541 }
2542
2543 static void control_get_public_ip_info(TALLOC_CTX *mem_ctx,
2544                                        struct tevent_req *req,
2545                                        struct ctdb_req_header *header,
2546                                        struct ctdb_req_control *request)
2547 {
2548         struct client_state *state = tevent_req_data(
2549                 req, struct client_state);
2550         struct ctdbd_context *ctdb = state->ctdb;
2551         struct ctdb_reply_control reply;
2552         ctdb_sock_addr *addr = request->rdata.data.addr;
2553         struct ctdb_public_ip_list *known = NULL;
2554         struct ctdb_public_ip_info *info = NULL;
2555         unsigned i;
2556
2557         reply.rdata.opcode = request->opcode;
2558
2559         info = talloc_zero(mem_ctx, struct ctdb_public_ip_info);
2560         if (info == NULL) {
2561                 reply.status = ENOMEM;
2562                 reply.errmsg = "Memory error";
2563                 goto done;
2564         }
2565
2566         reply.rdata.data.ipinfo = info;
2567
2568         if (ctdb->known_ips != NULL) {
2569                 known = &ctdb->known_ips[header->destnode];
2570         } else {
2571                 /* No IPs defined so create a dummy empty struct and
2572                  * fall through.  The given IP won't be matched
2573                  * below...
2574                  */
2575                 known = talloc_zero(mem_ctx, struct ctdb_public_ip_list);;
2576                 if (known == NULL) {
2577                         reply.status = ENOMEM;
2578                         reply.errmsg = "Memory error";
2579                         goto done;
2580                 }
2581         }
2582
2583         for (i = 0; i < known->num; i++) {
2584                 if (ctdb_sock_addr_same_ip(&known->ip[i].addr,
2585                                            addr)) {
2586                         break;
2587                 }
2588         }
2589
2590         if (i == known->num) {
2591                 D_ERR("GET_PUBLIC_IP_INFO: not known public IP %s\n",
2592                       ctdb_sock_addr_to_string(mem_ctx, addr, false));
2593                 reply.status = -1;
2594                 reply.errmsg = "Unknown address";
2595                 goto done;
2596         }
2597
2598         info->ip = known->ip[i];
2599
2600         /* The fake PUBLICIPS stanza and resulting known_ips data
2601          * don't know anything about interfaces, so completely fake
2602          * this.
2603          */
2604         info->active_idx = 0;
2605
2606         info->ifaces = get_ctdb_iface_list(mem_ctx, ctdb);
2607         if (info->ifaces == NULL) {
2608                 reply.status = ENOMEM;
2609                 reply.errmsg = "Memory error";
2610                 goto done;
2611         }
2612
2613         reply.status = 0;
2614         reply.errmsg = NULL;
2615
2616 done:
2617         client_send_control(req, header, &reply);
2618 }
2619
2620 static void control_get_ifaces(TALLOC_CTX *mem_ctx,
2621                                struct tevent_req *req,
2622                                struct ctdb_req_header *header,
2623                                struct ctdb_req_control *request)
2624 {
2625         struct client_state *state = tevent_req_data(
2626                 req, struct client_state);
2627         struct ctdbd_context *ctdb = state->ctdb;
2628         struct ctdb_reply_control reply;
2629         struct ctdb_iface_list *iface_list;
2630
2631         reply.rdata.opcode = request->opcode;
2632
2633         iface_list = get_ctdb_iface_list(mem_ctx, ctdb);
2634         if (iface_list == NULL) {
2635                 goto fail;
2636         }
2637
2638         reply.rdata.data.iface_list = iface_list;
2639         reply.status = 0;
2640         reply.errmsg = NULL;
2641         client_send_control(req, header, &reply);
2642         return;
2643
2644 fail:
2645         reply.status = -1;
2646         reply.errmsg = "Memory error";
2647         client_send_control(req, header, &reply);
2648 }
2649
2650 static void control_set_iface_link_state(TALLOC_CTX *mem_ctx,
2651                                          struct tevent_req *req,
2652                                          struct ctdb_req_header *header,
2653                                          struct ctdb_req_control *request)
2654 {
2655         struct client_state *state = tevent_req_data(
2656                 req, struct client_state);
2657         struct ctdbd_context *ctdb = state->ctdb;
2658         struct ctdb_reply_control reply;
2659         struct ctdb_iface *in_iface;
2660         struct interface *iface = NULL;
2661         bool link_up = false;
2662         int i;
2663
2664         reply.rdata.opcode = request->opcode;
2665
2666         in_iface = request->rdata.data.iface;
2667
2668         if (in_iface->name[CTDB_IFACE_SIZE] != '\0') {
2669                 reply.errmsg = "interface name not terminated";
2670                 goto fail;
2671         }
2672
2673         switch (in_iface->link_state) {
2674                 case 0:
2675                         link_up = false;
2676                         break;
2677
2678                 case 1:
2679                         link_up = true;
2680                         break;
2681
2682                 default:
2683                         reply.errmsg = "invalid link state";
2684                         goto fail;
2685         }
2686
2687         if (in_iface->references != 0) {
2688                 reply.errmsg = "references should be 0";
2689                 goto fail;
2690         }
2691
2692         for (i=0; i<ctdb->iface_map->num; i++) {
2693                 if (strcmp(ctdb->iface_map->iface[i].name,
2694                            in_iface->name) == 0) {
2695                         iface = &ctdb->iface_map->iface[i];
2696                         break;
2697                 }
2698         }
2699
2700         if (iface == NULL) {
2701                 reply.errmsg = "interface not found";
2702                 goto fail;
2703         }
2704
2705         iface->link_up = link_up;
2706
2707         reply.status = 0;
2708         reply.errmsg = NULL;
2709         client_send_control(req, header, &reply);
2710         return;
2711
2712 fail:
2713         reply.status = -1;
2714         client_send_control(req, header, &reply);
2715 }
2716
2717 static void control_set_db_readonly(TALLOC_CTX *mem_ctx,
2718                                     struct tevent_req *req,
2719                                     struct ctdb_req_header *header,
2720                                     struct ctdb_req_control *request)
2721 {
2722         struct client_state *state = tevent_req_data(
2723                 req, struct client_state);
2724         struct ctdbd_context *ctdb = state->ctdb;
2725         struct ctdb_reply_control reply;
2726         struct database *db;
2727
2728         reply.rdata.opcode = request->opcode;
2729
2730         db = database_find(ctdb->db_map, request->rdata.data.db_id);
2731         if (db == NULL) {
2732                 reply.status = ENOENT;
2733                 reply.errmsg = "Database not found";
2734                 goto done;
2735         }
2736
2737         if (db->flags & CTDB_DB_FLAGS_PERSISTENT) {
2738                 reply.status = EINVAL;
2739                 reply.errmsg = "Can not set READONLY on persistent db";
2740                 goto done;
2741         }
2742
2743         db->flags |= CTDB_DB_FLAGS_READONLY;
2744         reply.status = 0;
2745         reply.errmsg = NULL;
2746
2747 done:
2748         client_send_control(req, header, &reply);
2749 }
2750
2751 static void control_set_db_sticky(TALLOC_CTX *mem_ctx,
2752                                     struct tevent_req *req,
2753                                     struct ctdb_req_header *header,
2754                                     struct ctdb_req_control *request)
2755 {
2756         struct client_state *state = tevent_req_data(
2757                 req, struct client_state);
2758         struct ctdbd_context *ctdb = state->ctdb;
2759         struct ctdb_reply_control reply;
2760         struct database *db;
2761
2762         reply.rdata.opcode = request->opcode;
2763
2764         db = database_find(ctdb->db_map, request->rdata.data.db_id);
2765         if (db == NULL) {
2766                 reply.status = ENOENT;
2767                 reply.errmsg = "Database not found";
2768                 goto done;
2769         }
2770
2771         if (db->flags & CTDB_DB_FLAGS_PERSISTENT) {
2772                 reply.status = EINVAL;
2773                 reply.errmsg = "Can not set STICKY on persistent db";
2774                 goto done;
2775         }
2776
2777         db->flags |= CTDB_DB_FLAGS_STICKY;
2778         reply.status = 0;
2779         reply.errmsg = NULL;
2780
2781 done:
2782         client_send_control(req, header, &reply);
2783 }
2784
2785 static void control_ipreallocated(TALLOC_CTX *mem_ctx,
2786                                   struct tevent_req *req,
2787                                   struct ctdb_req_header *header,
2788                                   struct ctdb_req_control *request)
2789 {
2790         struct ctdb_reply_control reply;
2791
2792         /* Always succeed */
2793         reply.rdata.opcode = request->opcode;
2794         reply.status = 0;
2795         reply.errmsg = NULL;
2796
2797         client_send_control(req, header, &reply);
2798 }
2799
2800 static void control_get_runstate(TALLOC_CTX *mem_ctx,
2801                                  struct tevent_req *req,
2802                                  struct ctdb_req_header *header,
2803                                  struct ctdb_req_control *request)
2804 {
2805         struct client_state *state = tevent_req_data(
2806                 req, struct client_state);
2807         struct ctdbd_context *ctdb = state->ctdb;
2808         struct ctdb_reply_control reply;
2809
2810         reply.rdata.opcode = request->opcode;
2811         reply.rdata.data.runstate = ctdb->runstate;
2812         reply.status = 0;
2813         reply.errmsg = NULL;
2814
2815         client_send_control(req, header, &reply);
2816 }
2817
2818 static void control_get_nodes_file(TALLOC_CTX *mem_ctx,
2819                                    struct tevent_req *req,
2820                                    struct ctdb_req_header *header,
2821                                    struct ctdb_req_control *request)
2822 {
2823         struct ctdb_reply_control reply;
2824         struct ctdb_node_map *nodemap;
2825
2826         reply.rdata.opcode = request->opcode;
2827
2828         nodemap = read_nodes_file(mem_ctx, header->destnode);
2829         if (nodemap == NULL) {
2830                 goto fail;
2831         }
2832
2833         reply.rdata.data.nodemap = nodemap;
2834         reply.status = 0;
2835         reply.errmsg = NULL;
2836         client_send_control(req, header, &reply);
2837         return;
2838
2839 fail:
2840         reply.status = -1;
2841         reply.errmsg = "Failed to read nodes file";
2842         client_send_control(req, header, &reply);
2843 }
2844
2845 static void control_check_pid_srvid(TALLOC_CTX *mem_ctx,
2846                                     struct tevent_req *req,
2847                                     struct ctdb_req_header *header,
2848                                     struct ctdb_req_control *request)
2849 {
2850         struct client_state *state = tevent_req_data(
2851                 req, struct client_state);
2852         struct ctdbd_context *ctdb = state->ctdb;
2853         struct ctdb_client *client;
2854         struct client_state *cstate;
2855         struct ctdb_reply_control reply;
2856         bool pid_found, srvid_found;
2857         int ret;
2858
2859         reply.rdata.opcode = request->opcode;
2860
2861         pid_found = false;
2862         srvid_found = false;
2863
2864         for (client=ctdb->client_list; client != NULL; client=client->next) {
2865                 if (client->pid == request->rdata.data.pid_srvid->pid) {
2866                         pid_found = true;
2867                         cstate = (struct client_state *)client->state;
2868                         ret = srvid_exists(ctdb->srv,
2869                                            request->rdata.data.pid_srvid->srvid,
2870                                            cstate);
2871                         if (ret == 0) {
2872                                 srvid_found = true;
2873                                 ret = kill(cstate->pid, 0);
2874                                 if (ret != 0) {
2875                                         reply.status = ret;
2876                                         reply.errmsg = strerror(errno);
2877                                 } else {
2878                                         reply.status = 0;
2879                                         reply.errmsg = NULL;
2880                                 }
2881                         }
2882                 }
2883         }
2884
2885         if (! pid_found) {
2886                 reply.status = -1;
2887                 reply.errmsg = "No client for PID";
2888         } else if (! srvid_found) {
2889                 reply.status = -1;
2890                 reply.errmsg = "No client for PID and SRVID";
2891         }
2892
2893         client_send_control(req, header, &reply);
2894 }
2895
2896 static bool fake_control_failure(TALLOC_CTX *mem_ctx,
2897                                  struct tevent_req *req,
2898                                  struct ctdb_req_header *header,
2899                                  struct ctdb_req_control *request)
2900 {
2901         struct client_state *state = tevent_req_data(
2902                 req, struct client_state);
2903         struct ctdbd_context *ctdb = state->ctdb;
2904         struct ctdb_reply_control reply;
2905         struct fake_control_failure *f = NULL;
2906
2907         D_DEBUG("Checking fake control failure for control %u on node %u\n",
2908                 request->opcode, header->destnode);
2909         for (f = ctdb->control_failures; f != NULL; f = f->next) {
2910                 if (f->opcode == request->opcode &&
2911                     (f->pnn == header->destnode ||
2912                      f->pnn == CTDB_UNKNOWN_PNN)) {
2913
2914                         reply.rdata.opcode = request->opcode;
2915                         if (strcmp(f->error, "TIMEOUT") == 0) {
2916                                 /* Causes no reply */
2917                                 D_ERR("Control %u fake timeout on node %u\n",
2918                                       request->opcode, header->destnode);
2919                                 return true;
2920                         } else if (strcmp(f->error, "ERROR") == 0) {
2921                                 D_ERR("Control %u fake error on node %u\n",
2922                                       request->opcode, header->destnode);
2923                                 reply.status = -1;
2924                                 reply.errmsg = f->comment;
2925                                 client_send_control(req, header, &reply);
2926                                 return true;
2927                         }
2928                 }
2929         }
2930
2931         return false;
2932 }
2933
2934 static void control_error(TALLOC_CTX *mem_ctx,
2935                           struct tevent_req *req,
2936                           struct ctdb_req_header *header,
2937                           struct ctdb_req_control *request)
2938 {
2939         struct ctdb_reply_control reply;
2940
2941         reply.rdata.opcode = request->opcode;
2942         reply.status = -1;
2943         reply.errmsg = "Not implemented";
2944
2945         client_send_control(req, header, &reply);
2946 }
2947
2948 /*
2949  * Handling protocol - messages
2950  */
2951
2952 struct disable_recoveries_state {
2953         struct node *node;
2954 };
2955
2956 static void disable_recoveries_callback(struct tevent_req *subreq)
2957 {
2958         struct disable_recoveries_state *substate = tevent_req_callback_data(
2959                 subreq, struct disable_recoveries_state);
2960         bool status;
2961
2962         status = tevent_wakeup_recv(subreq);
2963         TALLOC_FREE(subreq);
2964         if (! status) {
2965                 DEBUG(DEBUG_INFO, ("tevent_wakeup_recv failed\n"));
2966         }
2967
2968         substate->node->recovery_disabled = false;
2969         TALLOC_FREE(substate->node->recovery_substate);
2970 }
2971
2972 static void message_disable_recoveries(TALLOC_CTX *mem_ctx,
2973                                        struct tevent_req *req,
2974                                        struct ctdb_req_header *header,
2975                                        struct ctdb_req_message *request)
2976 {
2977         struct client_state *state = tevent_req_data(
2978                 req, struct client_state);
2979         struct tevent_req *subreq;
2980         struct ctdbd_context *ctdb = state->ctdb;
2981         struct disable_recoveries_state *substate;
2982         struct ctdb_disable_message *disable = request->data.disable;
2983         struct ctdb_req_message_data reply;
2984         struct node *node;
2985         int ret = -1;
2986         TDB_DATA data;
2987
2988         node = &ctdb->node_map->node[header->destnode];
2989
2990         if (disable->timeout == 0) {
2991                 TALLOC_FREE(node->recovery_substate);
2992                 node->recovery_disabled = false;
2993                 DEBUG(DEBUG_INFO, ("Enabled recoveries on node %u\n",
2994                                    header->destnode));
2995                 goto done;
2996         }
2997
2998         substate = talloc_zero(ctdb->node_map,
2999                                struct disable_recoveries_state);
3000         if (substate == NULL) {
3001                 goto fail;
3002         }
3003
3004         substate->node = node;
3005
3006         subreq = tevent_wakeup_send(substate, state->ev,
3007                                     tevent_timeval_current_ofs(
3008                                             disable->timeout, 0));
3009         if (subreq == NULL) {
3010                 talloc_free(substate);
3011                 goto fail;
3012         }
3013         tevent_req_set_callback(subreq, disable_recoveries_callback, substate);
3014
3015         DEBUG(DEBUG_INFO, ("Disabled recoveries for %d seconds on node %u\n",
3016                            disable->timeout, header->destnode));
3017         node->recovery_substate = substate;
3018         node->recovery_disabled = true;
3019
3020 done:
3021         ret = header->destnode;
3022
3023 fail:
3024         reply.srvid = disable->srvid;
3025         data.dptr = (uint8_t *)&ret;
3026         data.dsize = sizeof(int);
3027         reply.data = data;
3028
3029         client_send_message(req, header, &reply);
3030 }
3031
3032 static void message_takeover_run(TALLOC_CTX *mem_ctx,
3033                                  struct tevent_req *req,
3034                                  struct ctdb_req_header *header,
3035                                  struct ctdb_req_message *request)
3036 {
3037         struct client_state *state = tevent_req_data(
3038                 req, struct client_state);
3039         struct ctdbd_context *ctdb = state->ctdb;
3040         struct ctdb_srvid_message *srvid = request->data.msg;
3041         struct ctdb_req_message_data reply;
3042         int ret = -1;
3043         TDB_DATA data;
3044
3045         if (header->destnode != ctdb->node_map->recmaster) {
3046                 /* No reply! Only recmaster replies... */
3047                 return;
3048         }
3049
3050         DEBUG(DEBUG_INFO, ("IP takover run on node %u\n",
3051                            header->destnode));
3052         ret = header->destnode;
3053
3054         reply.srvid = srvid->srvid;
3055         data.dptr = (uint8_t *)&ret;
3056         data.dsize = sizeof(int);
3057         reply.data = data;
3058
3059         client_send_message(req, header, &reply);
3060 }
3061
3062 /*
3063  * Handle a single client
3064  */
3065
3066 static void client_read_handler(uint8_t *buf, size_t buflen,
3067                                 void *private_data);
3068 static void client_dead_handler(void *private_data);
3069 static void client_process_packet(struct tevent_req *req,
3070                                   uint8_t *buf, size_t buflen);
3071 static void client_process_message(struct tevent_req *req,
3072                                    uint8_t *buf, size_t buflen);
3073 static void client_process_control(struct tevent_req *req,
3074                                    uint8_t *buf, size_t buflen);
3075 static void client_reply_done(struct tevent_req *subreq);
3076
3077 static struct tevent_req *client_send(TALLOC_CTX *mem_ctx,
3078                                       struct tevent_context *ev,
3079                                       int fd, struct ctdbd_context *ctdb,
3080                                       int pnn)
3081 {
3082         struct tevent_req *req;
3083         struct client_state *state;
3084         struct ucred cr;
3085         socklen_t crl = sizeof(struct ucred);
3086         int ret;
3087
3088         req = tevent_req_create(mem_ctx, &state, struct client_state);
3089         if (req == NULL) {
3090                 return NULL;
3091         }
3092
3093         state->ev = ev;
3094         state->fd = fd;
3095         state->ctdb = ctdb;
3096         state->pnn = pnn;
3097
3098         ret = getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &cr, &crl);
3099         if (ret != 0) {
3100                 tevent_req_error(req, ret);
3101                 return tevent_req_post(req, ev);
3102         }
3103         state->pid = cr.pid;
3104
3105         ret = comm_setup(state, ev, fd, client_read_handler, req,
3106                          client_dead_handler, req, &state->comm);
3107         if (ret != 0) {
3108                 tevent_req_error(req, ret);
3109                 return tevent_req_post(req, ev);
3110         }
3111
3112         ret = client_add(ctdb, state->pid, state);
3113         if (ret != 0) {
3114                 tevent_req_error(req, ret);
3115                 return tevent_req_post(req, ev);
3116         }
3117
3118         DEBUG(DEBUG_INFO, ("New client fd=%d\n", fd));
3119
3120         return req;
3121 }
3122
3123 static void client_read_handler(uint8_t *buf, size_t buflen,
3124                                 void *private_data)
3125 {
3126         struct tevent_req *req = talloc_get_type_abort(
3127                 private_data, struct tevent_req);
3128         struct client_state *state = tevent_req_data(
3129                 req, struct client_state);
3130         struct ctdbd_context *ctdb = state->ctdb;
3131         struct ctdb_req_header header;
3132         size_t np;
3133         int ret, i;
3134
3135         ret = ctdb_req_header_pull(buf, buflen, &header, &np);
3136         if (ret != 0) {
3137                 return;
3138         }
3139
3140         if (buflen != header.length) {
3141                 return;
3142         }
3143
3144         ret = ctdb_req_header_verify(&header, 0);
3145         if (ret != 0) {
3146                 return;
3147         }
3148
3149         header_fix_pnn(&header, ctdb);
3150
3151         if (header.destnode == CTDB_BROADCAST_ALL) {
3152                 for (i=0; i<ctdb->node_map->num_nodes; i++) {
3153                         header.destnode = i;
3154
3155                         ctdb_req_header_push(&header, buf, &np);
3156                         client_process_packet(req, buf, buflen);
3157                 }
3158                 return;
3159         }
3160
3161         if (header.destnode == CTDB_BROADCAST_CONNECTED) {
3162                 for (i=0; i<ctdb->node_map->num_nodes; i++) {
3163                         if (ctdb->node_map->node[i].flags &
3164                             NODE_FLAGS_DISCONNECTED) {
3165                                 continue;
3166                         }
3167
3168                         header.destnode = i;
3169
3170                         ctdb_req_header_push(&header, buf, &np);
3171                         client_process_packet(req, buf, buflen);
3172                 }
3173                 return;
3174         }
3175
3176         if (header.destnode > ctdb->node_map->num_nodes) {
3177                 fprintf(stderr, "Invalid destination pnn 0x%x\n",
3178                         header.destnode);
3179                 return;
3180         }
3181
3182
3183         if (ctdb->node_map->node[header.destnode].flags & NODE_FLAGS_DISCONNECTED) {
3184                 fprintf(stderr, "Packet for disconnected node pnn %u\n",
3185                         header.destnode);
3186                 return;
3187         }
3188
3189         ctdb_req_header_push(&header, buf, &np);
3190         client_process_packet(req, buf, buflen);
3191 }
3192
3193 static void client_dead_handler(void *private_data)
3194 {
3195         struct tevent_req *req = talloc_get_type_abort(
3196                 private_data, struct tevent_req);
3197
3198         tevent_req_done(req);
3199 }
3200
3201 static void client_process_packet(struct tevent_req *req,
3202                                   uint8_t *buf, size_t buflen)
3203 {
3204         struct ctdb_req_header header;
3205         size_t np;
3206         int ret;
3207
3208         ret = ctdb_req_header_pull(buf, buflen, &header, &np);
3209         if (ret != 0) {
3210                 return;
3211         }
3212
3213         switch (header.operation) {
3214         case CTDB_REQ_MESSAGE:
3215                 client_process_message(req, buf, buflen);
3216                 break;
3217
3218         case CTDB_REQ_CONTROL:
3219                 client_process_control(req, buf, buflen);
3220                 break;
3221
3222         default:
3223                 break;
3224         }
3225 }
3226
3227 static void client_process_message(struct tevent_req *req,
3228                                    uint8_t *buf, size_t buflen)
3229 {
3230         struct client_state *state = tevent_req_data(
3231                 req, struct client_state);
3232         struct ctdbd_context *ctdb = state->ctdb;
3233         TALLOC_CTX *mem_ctx;
3234         struct ctdb_req_header header;
3235         struct ctdb_req_message request;
3236         uint64_t srvid;
3237         int ret;
3238
3239         mem_ctx = talloc_new(state);
3240         if (tevent_req_nomem(mem_ctx, req)) {
3241                 return;
3242         }
3243
3244         ret = ctdb_req_message_pull(buf, buflen, &header, mem_ctx, &request);
3245         if (ret != 0) {
3246                 talloc_free(mem_ctx);
3247                 tevent_req_error(req, ret);
3248                 return;
3249         }
3250
3251         header_fix_pnn(&header, ctdb);
3252
3253         if (header.destnode >= ctdb->node_map->num_nodes) {
3254                 /* Many messages are not replied to, so just behave as
3255                  * though this message was not received */
3256                 fprintf(stderr, "Invalid node %d\n", header.destnode);
3257                 talloc_free(mem_ctx);
3258                 return;
3259         }
3260
3261         srvid = request.srvid;
3262         DEBUG(DEBUG_INFO, ("request srvid = 0x%"PRIx64"\n", srvid));
3263
3264         if (srvid == CTDB_SRVID_DISABLE_RECOVERIES) {
3265                 message_disable_recoveries(mem_ctx, req, &header, &request);
3266         } else if (srvid == CTDB_SRVID_TAKEOVER_RUN) {
3267                 message_takeover_run(mem_ctx, req, &header, &request);
3268         }
3269
3270         /* check srvid */
3271         talloc_free(mem_ctx);
3272 }
3273
3274 static void client_process_control(struct tevent_req *req,
3275                                    uint8_t *buf, size_t buflen)
3276 {
3277         struct client_state *state = tevent_req_data(
3278                 req, struct client_state);
3279         struct ctdbd_context *ctdb = state->ctdb;
3280         TALLOC_CTX *mem_ctx;
3281         struct ctdb_req_header header;
3282         struct ctdb_req_control request;
3283         int ret;
3284
3285         mem_ctx = talloc_new(state);
3286         if (tevent_req_nomem(mem_ctx, req)) {
3287                 return;
3288         }
3289
3290         ret = ctdb_req_control_pull(buf, buflen, &header, mem_ctx, &request);
3291         if (ret != 0) {
3292                 talloc_free(mem_ctx);
3293                 tevent_req_error(req, ret);
3294                 return;
3295         }
3296
3297         header_fix_pnn(&header, ctdb);
3298
3299         if (header.destnode >= ctdb->node_map->num_nodes) {
3300                 struct ctdb_reply_control reply;
3301
3302                 reply.rdata.opcode = request.opcode;
3303                 reply.errmsg = "Invalid node";
3304                 reply.status = -1;
3305                 client_send_control(req, &header, &reply);
3306                 return;
3307         }
3308
3309         DEBUG(DEBUG_INFO, ("request opcode = %u, reqid = %u\n",
3310                            request.opcode, header.reqid));
3311
3312         if (fake_control_failure(mem_ctx, req, &header, &request)) {
3313                 goto done;
3314         }
3315
3316         switch (request.opcode) {
3317         case CTDB_CONTROL_PROCESS_EXISTS:
3318                 control_process_exists(mem_ctx, req, &header, &request);
3319                 break;
3320
3321         case CTDB_CONTROL_PING:
3322                 control_ping(mem_ctx, req, &header, &request);
3323                 break;
3324
3325         case CTDB_CONTROL_GETDBPATH:
3326                 control_getdbpath(mem_ctx, req, &header, &request);
3327                 break;
3328
3329         case CTDB_CONTROL_GETVNNMAP:
3330                 control_getvnnmap(mem_ctx, req, &header, &request);
3331                 break;
3332
3333         case CTDB_CONTROL_GET_DEBUG:
3334                 control_get_debug(mem_ctx, req, &header, &request);
3335                 break;
3336
3337         case CTDB_CONTROL_SET_DEBUG:
3338                 control_set_debug(mem_ctx, req, &header, &request);
3339                 break;
3340
3341         case CTDB_CONTROL_GET_DBMAP:
3342                 control_get_dbmap(mem_ctx, req, &header, &request);
3343                 break;
3344
3345         case CTDB_CONTROL_GET_RECMODE:
3346                 control_get_recmode(mem_ctx, req, &header, &request);
3347                 break;
3348
3349         case CTDB_CONTROL_SET_RECMODE:
3350                 control_set_recmode(mem_ctx, req, &header, &request);
3351                 break;
3352
3353         case CTDB_CONTROL_REGISTER_SRVID:
3354                 control_register_srvid(mem_ctx, req, &header, &request);
3355                 break;
3356
3357         case CTDB_CONTROL_DEREGISTER_SRVID:
3358                 control_deregister_srvid(mem_ctx, req, &header, &request);
3359                 break;
3360
3361         case CTDB_CONTROL_GET_DBNAME:
3362                 control_get_dbname(mem_ctx, req, &header, &request);
3363                 break;
3364
3365         case CTDB_CONTROL_GET_PID:
3366                 control_get_pid(mem_ctx, req, &header, &request);
3367                 break;
3368
3369         case CTDB_CONTROL_GET_RECMASTER:
3370                 control_get_recmaster(mem_ctx, req, &header, &request);
3371                 break;
3372
3373         case CTDB_CONTROL_GET_PNN:
3374                 control_get_pnn(mem_ctx, req, &header, &request);
3375                 break;
3376
3377         case CTDB_CONTROL_SHUTDOWN:
3378                 control_shutdown(mem_ctx, req, &header, &request);
3379                 break;
3380
3381         case CTDB_CONTROL_SET_TUNABLE:
3382                 control_set_tunable(mem_ctx, req, &header, &request);
3383                 break;
3384
3385         case CTDB_CONTROL_GET_TUNABLE:
3386                 control_get_tunable(mem_ctx, req, &header, &request);
3387                 break;
3388
3389         case CTDB_CONTROL_LIST_TUNABLES:
3390                 control_list_tunables(mem_ctx, req, &header, &request);
3391                 break;
3392
3393         case CTDB_CONTROL_MODIFY_FLAGS:
3394                 control_modify_flags(mem_ctx, req, &header, &request);
3395                 break;
3396
3397         case CTDB_CONTROL_GET_ALL_TUNABLES:
3398                 control_get_all_tunables(mem_ctx, req, &header, &request);
3399                 break;
3400
3401         case CTDB_CONTROL_UPTIME:
3402                 control_uptime(mem_ctx, req, &header, &request);
3403                 break;
3404
3405         case CTDB_CONTROL_RELOAD_NODES_FILE:
3406                 control_reload_nodes_file(mem_ctx, req, &header, &request);
3407                 break;
3408
3409         case CTDB_CONTROL_GET_CAPABILITIES:
3410                 control_get_capabilities(mem_ctx, req, &header, &request);
3411                 break;
3412
3413         case CTDB_CONTROL_RELEASE_IP:
3414                 control_release_ip(mem_ctx, req, &header, &request);
3415                 break;
3416
3417         case CTDB_CONTROL_TAKEOVER_IP:
3418                 control_takeover_ip(mem_ctx, req, &header, &request);
3419                 break;
3420
3421         case CTDB_CONTROL_GET_PUBLIC_IPS:
3422                 control_get_public_ips(mem_ctx, req, &header, &request);
3423                 break;
3424
3425         case CTDB_CONTROL_GET_NODEMAP:
3426                 control_get_nodemap(mem_ctx, req, &header, &request);
3427                 break;
3428
3429         case CTDB_CONTROL_GET_RECLOCK_FILE:
3430                 control_get_reclock_file(mem_ctx, req, &header, &request);
3431                 break;
3432
3433         case CTDB_CONTROL_STOP_NODE:
3434                 control_stop_node(mem_ctx, req, &header, &request);
3435                 break;
3436
3437         case CTDB_CONTROL_CONTINUE_NODE:
3438                 control_continue_node(mem_ctx, req, &header, &request);
3439                 break;
3440
3441         case CTDB_CONTROL_SET_BAN_STATE:
3442                 control_set_ban_state(mem_ctx, req, &header, &request);
3443                 break;
3444
3445         case CTDB_CONTROL_GET_DB_SEQNUM:
3446                 control_get_db_seqnum(mem_ctx, req, &header, &request);
3447                 break;
3448
3449         case CTDB_CONTROL_DB_GET_HEALTH:
3450                 control_db_get_health(mem_ctx, req, &header, &request);
3451                 break;
3452
3453         case CTDB_CONTROL_GET_PUBLIC_IP_INFO:
3454                 control_get_public_ip_info(mem_ctx, req, &header, &request);
3455                 break;
3456
3457         case CTDB_CONTROL_GET_IFACES:
3458                 control_get_ifaces(mem_ctx, req, &header, &request);
3459                 break;
3460
3461         case CTDB_CONTROL_SET_IFACE_LINK_STATE:
3462                 control_set_iface_link_state(mem_ctx, req, &header, &request);
3463                 break;
3464
3465         case CTDB_CONTROL_SET_DB_READONLY:
3466                 control_set_db_readonly(mem_ctx, req, &header, &request);
3467                 break;
3468
3469         case CTDB_CONTROL_SET_DB_STICKY:
3470                 control_set_db_sticky(mem_ctx, req, &header, &request);
3471                 break;
3472
3473         case CTDB_CONTROL_IPREALLOCATED:
3474                 control_ipreallocated(mem_ctx, req, &header, &request);
3475                 break;
3476
3477         case CTDB_CONTROL_GET_RUNSTATE:
3478                 control_get_runstate(mem_ctx, req, &header, &request);
3479                 break;
3480
3481         case CTDB_CONTROL_GET_NODES_FILE:
3482                 control_get_nodes_file(mem_ctx, req, &header, &request);
3483                 break;
3484
3485         case CTDB_CONTROL_CHECK_PID_SRVID:
3486                 control_check_pid_srvid(mem_ctx, req, &header, &request);
3487                 break;
3488
3489         default:
3490                 if (! (request.flags & CTDB_CTRL_FLAG_NOREPLY)) {
3491                         control_error(mem_ctx, req, &header, &request);
3492                 }
3493                 break;
3494         }
3495
3496 done:
3497         talloc_free(mem_ctx);
3498 }
3499
3500 static int client_recv(struct tevent_req *req, int *perr)
3501 {
3502         struct client_state *state = tevent_req_data(
3503                 req, struct client_state);
3504         int err;
3505
3506         DEBUG(DEBUG_INFO, ("Client done fd=%d\n", state->fd));
3507         close(state->fd);
3508
3509         if (tevent_req_is_unix_error(req, &err)) {
3510                 if (perr != NULL) {
3511                         *perr = err;
3512                 }
3513                 return -1;
3514         }
3515
3516         return state->status;
3517 }
3518
3519 /*
3520  * Fake CTDB server
3521  */
3522
3523 struct server_state {
3524         struct tevent_context *ev;
3525         struct ctdbd_context *ctdb;
3526         int fd;
3527 };
3528
3529 static void server_new_client(struct tevent_req *subreq);
3530 static void server_client_done(struct tevent_req *subreq);
3531
3532 static struct tevent_req *server_send(TALLOC_CTX *mem_ctx,
3533                                       struct tevent_context *ev,
3534                                       struct ctdbd_context *ctdb,
3535                                       int fd)
3536 {
3537         struct tevent_req *req, *subreq;
3538         struct server_state *state;
3539
3540         req = tevent_req_create(mem_ctx, &state, struct server_state);
3541         if (req == NULL) {
3542                 return NULL;
3543         }
3544
3545         state->ev = ev;
3546         state->ctdb = ctdb;
3547         state->fd = fd;
3548
3549         subreq = accept_send(state, ev, fd);
3550         if (tevent_req_nomem(subreq, req)) {
3551                 return tevent_req_post(req, ev);
3552         }
3553         tevent_req_set_callback(subreq, server_new_client, req);
3554
3555         return req;
3556 }
3557
3558 static void server_new_client(struct tevent_req *subreq)
3559 {
3560         struct tevent_req *req = tevent_req_callback_data(
3561                 subreq, struct tevent_req);
3562         struct server_state *state = tevent_req_data(
3563                 req, struct server_state);
3564         struct ctdbd_context *ctdb = state->ctdb;
3565         int client_fd;
3566         int ret = 0;
3567
3568         client_fd = accept_recv(subreq, NULL, NULL, &ret);
3569         TALLOC_FREE(subreq);
3570         if (client_fd == -1) {
3571                 tevent_req_error(req, ret);
3572                 return;
3573         }
3574
3575         subreq = client_send(state, state->ev, client_fd,
3576                              ctdb, ctdb->node_map->pnn);
3577         if (tevent_req_nomem(subreq, req)) {
3578                 return;
3579         }
3580         tevent_req_set_callback(subreq, server_client_done, req);
3581
3582         ctdb->num_clients += 1;
3583
3584         subreq = accept_send(state, state->ev, state->fd);
3585         if (tevent_req_nomem(subreq, req)) {
3586                 return;
3587         }
3588         tevent_req_set_callback(subreq, server_new_client, req);
3589 }
3590
3591 static void server_client_done(struct tevent_req *subreq)
3592 {
3593         struct tevent_req *req = tevent_req_callback_data(
3594                 subreq, struct tevent_req);
3595         struct server_state *state = tevent_req_data(
3596                 req, struct server_state);
3597         struct ctdbd_context *ctdb = state->ctdb;
3598         int ret = 0;
3599         int status;
3600
3601         status = client_recv(subreq, &ret);
3602         TALLOC_FREE(subreq);
3603         if (status < 0) {
3604                 tevent_req_error(req, ret);
3605                 return;
3606         }
3607
3608         ctdb->num_clients -= 1;
3609
3610         if (status == 99) {
3611                 /* Special status, to shutdown server */
3612                 DEBUG(DEBUG_INFO, ("Shutting down server\n"));
3613                 tevent_req_done(req);
3614         }
3615 }
3616
3617 static bool server_recv(struct tevent_req *req, int *perr)
3618 {
3619         int err;
3620
3621         if (tevent_req_is_unix_error(req, &err)) {
3622                 if (perr != NULL) {
3623                         *perr = err;
3624                 }
3625                 return false;
3626         }
3627         return true;
3628 }
3629
3630 /*
3631  * Main functions
3632  */
3633
3634 static int socket_init(const char *sockpath)
3635 {
3636         struct sockaddr_un addr;
3637         size_t len;
3638         int ret, fd;
3639
3640         memset(&addr, 0, sizeof(addr));
3641         addr.sun_family = AF_UNIX;
3642
3643         len = strlcpy(addr.sun_path, sockpath, sizeof(addr.sun_path));
3644         if (len >= sizeof(addr.sun_path)) {
3645                 fprintf(stderr, "path too long: %s\n", sockpath);
3646                 return -1;
3647         }
3648
3649         fd = socket(AF_UNIX, SOCK_STREAM, 0);
3650         if (fd == -1) {
3651                 fprintf(stderr, "socket failed - %s\n", sockpath);
3652                 return -1;
3653         }
3654
3655         ret = bind(fd, (struct sockaddr *)&addr, sizeof(addr));
3656         if (ret != 0) {
3657                 fprintf(stderr, "bind failed - %s\n", sockpath);
3658                 goto fail;
3659         }
3660
3661         ret = listen(fd, 10);
3662         if (ret != 0) {
3663                 fprintf(stderr, "listen failed\n");
3664                 goto fail;
3665         }
3666
3667         DEBUG(DEBUG_INFO, ("Socket init done\n"));
3668
3669         return fd;
3670
3671 fail:
3672         if (fd != -1) {
3673                 close(fd);
3674         }
3675         return -1;
3676 }
3677
3678 static struct options {
3679         const char *sockpath;
3680         const char *pidfile;
3681         const char *debuglevel;
3682 } options;
3683
3684 static struct poptOption cmdline_options[] = {
3685         { "socket", 's', POPT_ARG_STRING, &options.sockpath, 0,
3686                 "Unix domain socket path", "filename" },
3687         { "pidfile", 'p', POPT_ARG_STRING, &options.pidfile, 0,
3688                 "pid file", "filename" } ,
3689         { "debug", 'd', POPT_ARG_STRING, &options.debuglevel, 0,
3690                 "debug level", "ERR|WARNING|NOTICE|INFO|DEBUG" } ,
3691 };
3692
3693 static void cleanup(void)
3694 {
3695         unlink(options.sockpath);
3696         unlink(options.pidfile);
3697 }
3698
3699 static void signal_handler(int sig)
3700 {
3701         cleanup();
3702         exit(0);
3703 }
3704
3705 static void start_server(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
3706                          struct ctdbd_context *ctdb, int fd, int pfd)
3707 {
3708         struct tevent_req *req;
3709         int ret = 0;
3710         ssize_t len;
3711
3712         atexit(cleanup);
3713         signal(SIGTERM, signal_handler);
3714
3715         req = server_send(mem_ctx, ev, ctdb, fd);
3716         if (req == NULL) {
3717                 fprintf(stderr, "Memory error\n");
3718                 exit(1);
3719         }
3720
3721         len = write(pfd, &ret, sizeof(ret));
3722         if (len != sizeof(ret)) {
3723                 fprintf(stderr, "Failed to send message to parent\n");
3724                 exit(1);
3725         }
3726         close(pfd);
3727
3728         tevent_req_poll(req, ev);
3729
3730         server_recv(req, &ret);
3731         if (ret != 0) {
3732                 exit(1);
3733         }
3734 }
3735
3736 int main(int argc, const char *argv[])
3737 {
3738         TALLOC_CTX *mem_ctx;
3739         struct ctdbd_context *ctdb;
3740         struct tevent_context *ev;
3741         poptContext pc;
3742         int opt, fd, ret, pfd[2];
3743         ssize_t len;
3744         pid_t pid;
3745         FILE *fp;
3746
3747         pc = poptGetContext(argv[0], argc, argv, cmdline_options,
3748                             POPT_CONTEXT_KEEP_FIRST);
3749         while ((opt = poptGetNextOpt(pc)) != -1) {
3750                 fprintf(stderr, "Invalid option %s\n", poptBadOption(pc, 0));
3751                 exit(1);
3752         }
3753
3754         if (options.sockpath == NULL) {
3755                 fprintf(stderr, "Please specify socket path\n");
3756                 poptPrintHelp(pc, stdout, 0);
3757                 exit(1);
3758         }
3759
3760         if (options.pidfile == NULL) {
3761                 fprintf(stderr, "Please specify pid file\n");
3762                 poptPrintHelp(pc, stdout, 0);
3763                 exit(1);
3764         }
3765
3766         mem_ctx = talloc_new(NULL);
3767         if (mem_ctx == NULL) {
3768                 fprintf(stderr, "Memory error\n");
3769                 exit(1);
3770         }
3771
3772         ret = logging_init(mem_ctx, "file:", options.debuglevel, "fake-ctdbd");
3773         if (ret != 0) {
3774                 fprintf(stderr, "Invalid debug level\n");
3775                 poptPrintHelp(pc, stdout, 0);
3776                 exit(1);
3777         }
3778
3779         ctdb = ctdbd_setup(mem_ctx);
3780         if (ctdb == NULL) {
3781                 exit(1);
3782         }
3783
3784         if (! ctdbd_verify(ctdb)) {
3785                 exit(1);
3786         }
3787
3788         ev = tevent_context_init(mem_ctx);
3789         if (ev == NULL) {
3790                 fprintf(stderr, "Memory error\n");
3791                 exit(1);
3792         }
3793
3794         fd = socket_init(options.sockpath);
3795         if (fd == -1) {
3796                 exit(1);
3797         }
3798
3799         ret = pipe(pfd);
3800         if (ret != 0) {
3801                 fprintf(stderr, "Failed to create pipe\n");
3802                 cleanup();
3803                 exit(1);
3804         }
3805
3806         pid = fork();
3807         if (pid == -1) {
3808                 fprintf(stderr, "Failed to fork\n");
3809                 cleanup();
3810                 exit(1);
3811         }
3812
3813         if (pid == 0) {
3814                 /* Child */
3815                 close(pfd[0]);
3816                 start_server(mem_ctx, ev, ctdb, fd, pfd[1]);
3817                 exit(1);
3818         }
3819
3820         /* Parent */
3821         close(pfd[1]);
3822
3823         len = read(pfd[0], &ret, sizeof(ret));
3824         close(pfd[0]);
3825         if (len != sizeof(ret)) {
3826                 fprintf(stderr, "len = %zi\n", len);
3827                 fprintf(stderr, "Failed to get message from child\n");
3828                 kill(pid, SIGTERM);
3829                 exit(1);
3830         }
3831
3832         fp = fopen(options.pidfile, "w");
3833         if (fp == NULL) {
3834                 fprintf(stderr, "Failed to open pid file %s\n",
3835                         options.pidfile);
3836                 kill(pid, SIGTERM);
3837                 exit(1);
3838         }
3839         fprintf(fp, "%d\n", pid);
3840         fclose(fp);
3841
3842         return 0;
3843 }