ctdb/tools/ctdb.c

   1 /*
   2    CTDB control tool
   3
   4    Copyright (C) Amitay Isaacs  2015
   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/filesys.h"
  23 #include "system/time.h"
  24 #include "system/wait.h"
  25 #include "system/dir.h"
  26
  27 #include <ctype.h>
  28 #include <popt.h>
  29 #include <talloc.h>
  30 #include <tevent.h>
  31 #include <tdb.h>
  32
  33 #include "ctdb_version.h"
  34 #include "lib/util/debug.h"
  35 #include "lib/util/samba_util.h"
  36 #include "lib/util/sys_rw.h"
  37
  38 #include "common/db_hash.h"
  39 #include "common/logging.h"
  40 #include "protocol/protocol.h"
  41 #include "protocol/protocol_api.h"
  42 #include "protocol/protocol_util.h"
  43 #include "common/system.h"
  44 #include "client/client.h"
  45 #include "client/client_sync.h"
  46
  47 #define TIMEOUT()       timeval_current_ofs(options.timelimit, 0)
  48
  49 #define SRVID_CTDB_TOOL    (CTDB_SRVID_TOOL_RANGE | 0x0001000000000000LL)
  50 #define SRVID_CTDB_PUSHDB  (CTDB_SRVID_TOOL_RANGE | 0x0002000000000000LL)
  51
  52 static struct {
  53         const char *socket;
  54         const char *debuglevelstr;
  55         int timelimit;
  56         int pnn;
  57         int machinereadable;
  58         const char *sep;
  59         int machineparsable;
  60         int verbose;
  61         int maxruntime;
  62         int printemptyrecords;
  63         int printdatasize;
  64         int printlmaster;
  65         int printhash;
  66         int printrecordflags;
  67 } options;
  68
  69 static poptContext pc;
  70
  71 struct ctdb_context {
  72         struct tevent_context *ev;
  73         struct ctdb_client_context *client;
  74         struct ctdb_node_map *nodemap;
  75         uint32_t pnn, cmd_pnn;
  76         uint64_t srvid;
  77 };
  78
  79 static void usage(const char *command);
  80
  81 /*
  82  * Utility Functions
  83  */
  84
  85 static double timeval_delta(struct timeval *tv2, struct timeval *tv)
  86 {
  87         return (tv2->tv_sec - tv->tv_sec) +
  88                (tv2->tv_usec - tv->tv_usec) * 1.0e-6;
  89 }
  90
  91 static struct ctdb_node_and_flags *get_node_by_pnn(
  92                                         struct ctdb_node_map *nodemap,
  93                                         uint32_t pnn)
  94 {
  95         int i;
  96
  97         for (i=0; i<nodemap->num; i++) {
  98                 if (nodemap->node[i].pnn == pnn) {
  99                         return &nodemap->node[i];
 100                 }
 101         }
 102         return NULL;
 103 }
 104
 105 static const char *pretty_print_flags(TALLOC_CTX *mem_ctx, uint32_t flags)
 106 {
 107         static const struct {
 108                 uint32_t flag;
 109                 const char *name;
 110         } flag_names[] = {
 111                 { NODE_FLAGS_DISCONNECTED,          "DISCONNECTED" },
 112                 { NODE_FLAGS_PERMANENTLY_DISABLED,  "DISABLED" },
 113                 { NODE_FLAGS_BANNED,                "BANNED" },
 114                 { NODE_FLAGS_UNHEALTHY,             "UNHEALTHY" },
 115                 { NODE_FLAGS_DELETED,               "DELETED" },
 116                 { NODE_FLAGS_STOPPED,               "STOPPED" },
 117                 { NODE_FLAGS_INACTIVE,              "INACTIVE" },
 118         };
 119         char *flags_str = NULL;
 120         int i;
 121
 122         for (i=0; i<ARRAY_SIZE(flag_names); i++) {
 123                 if (flags & flag_names[i].flag) {
 124                         if (flags_str == NULL) {
 125                                 flags_str = talloc_asprintf(mem_ctx,
 126                                                 "%s", flag_names[i].name);
 127                         } else {
 128                                 flags_str = talloc_asprintf_append(flags_str,
 129                                                 "|%s", flag_names[i].name);
 130                         }
 131                         if (flags_str == NULL) {
 132                                 return "OUT-OF-MEMORY";
 133                         }
 134                 }
 135         }
 136         if (flags_str == NULL) {
 137                 return "OK";
 138         }
 139
 140         return flags_str;
 141 }
 142
 143 static uint64_t next_srvid(struct ctdb_context *ctdb)
 144 {
 145         ctdb->srvid += 1;
 146         return ctdb->srvid;
 147 }
 148
 149 /*
 150  * Get consistent nodemap information.
 151  *
 152  * If nodemap is already cached, use that. If not get it.
 153  * If the current node is BANNED, then get nodemap from "better" node.
 154  */
 155 static struct ctdb_node_map *get_nodemap(struct ctdb_context *ctdb, bool force)
 156 {
 157         TALLOC_CTX *tmp_ctx;
 158         struct ctdb_node_map *nodemap;
 159         struct ctdb_node_and_flags *node;
 160         uint32_t current_node;
 161         int ret;
 162
 163         if (force) {
 164                 TALLOC_FREE(ctdb->nodemap);
 165         }
 166
 167         if (ctdb->nodemap != NULL) {
 168                 return ctdb->nodemap;
 169         }
 170
 171         tmp_ctx = talloc_new(ctdb);
 172         if (tmp_ctx == NULL) {
 173                 return false;
 174         }
 175
 176         current_node = ctdb->pnn;
 177 again:
 178         ret = ctdb_ctrl_get_nodemap(tmp_ctx, ctdb->ev, ctdb->client,
 179                                     current_node, TIMEOUT(), &nodemap);
 180         if (ret != 0) {
 181                 fprintf(stderr, "Failed to get nodemap from node %u\n",
 182                         current_node);
 183                 goto failed;
 184         }
 185
 186         node = get_node_by_pnn(nodemap, current_node);
 187         if (node->flags & NODE_FLAGS_BANNED) {
 188                 /* Pick next node */
 189                 do {
 190                         current_node = (current_node + 1) % nodemap->num;
 191                         node = get_node_by_pnn(nodemap, current_node);
 192                         if (! (node->flags &
 193                               (NODE_FLAGS_DELETED|NODE_FLAGS_DISCONNECTED))) {
 194                                 break;
 195                         }
 196                 } while (current_node != ctdb->pnn);
 197
 198                 if (current_node == ctdb->pnn) {
 199                         /* Tried all nodes in the cluster */
 200                         fprintf(stderr, "Warning: All nodes are banned.\n");
 201                         goto failed;
 202                 }
 203
 204                 goto again;
 205         }
 206
 207         ctdb->nodemap = talloc_steal(ctdb, nodemap);
 208         return nodemap;
 209
 210 failed:
 211         talloc_free(tmp_ctx);
 212         return NULL;
 213 }
 214
 215 static bool verify_pnn(struct ctdb_context *ctdb, int pnn)
 216 {
 217         struct ctdb_node_map *nodemap;
 218         bool found;
 219         int i;
 220
 221         if (pnn == -1) {
 222                 return false;
 223         }
 224
 225         nodemap = get_nodemap(ctdb, false);
 226         if (nodemap == NULL) {
 227                 return false;
 228         }
 229
 230         found = false;
 231         for (i=0; i<nodemap->num; i++) {
 232                 if (nodemap->node[i].pnn == pnn) {
 233                         found = true;
 234                         break;
 235                 }
 236         }
 237         if (! found) {
 238                 fprintf(stderr, "Node %u does not exist\n", pnn);
 239                 return false;
 240         }
 241
 242         if (nodemap->node[i].flags &
 243             (NODE_FLAGS_DISCONNECTED|NODE_FLAGS_DELETED)) {
 244                 fprintf(stderr, "Node %u has status %s\n", pnn,
 245                         pretty_print_flags(ctdb, nodemap->node[i].flags));
 246                 return false;
 247         }
 248
 249         return true;
 250 }
 251
 252 static struct ctdb_node_map *talloc_nodemap(TALLOC_CTX *mem_ctx,
 253                                             struct ctdb_node_map *nodemap)
 254 {
 255         struct ctdb_node_map *nodemap2;
 256
 257         nodemap2 = talloc_zero(mem_ctx, struct ctdb_node_map);
 258         if (nodemap2 == NULL) {
 259                 return NULL;
 260         }
 261
 262         nodemap2->node = talloc_array(nodemap2, struct ctdb_node_and_flags,
 263                                       nodemap->num);
 264         if (nodemap2->node == NULL) {
 265                 talloc_free(nodemap2);
 266                 return NULL;
 267         }
 268
 269         return nodemap2;
 270 }
 271
 272 /*
 273  * Get the number and the list of matching nodes
 274  *
 275  *   nodestring :=  NULL | all | pnn,[pnn,...]
 276  *
 277  * If nodestring is NULL, use the current node.
 278  */
 279 static bool parse_nodestring(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 280                              const char *nodestring,
 281                              struct ctdb_node_map **out)
 282 {
 283         struct ctdb_node_map *nodemap, *nodemap2;
 284         struct ctdb_node_and_flags *node;
 285         int i;
 286
 287         nodemap = get_nodemap(ctdb, false);
 288         if (nodemap == NULL) {
 289                 return false;
 290         }
 291
 292         nodemap2 = talloc_nodemap(mem_ctx, nodemap);
 293         if (nodemap2 == NULL) {
 294                 return false;
 295         }
 296
 297         if (nodestring == NULL) {
 298                 for (i=0; i<nodemap->num; i++) {
 299                         if (nodemap->node[i].pnn == ctdb->cmd_pnn) {
 300                                 nodemap2->node[0] = nodemap->node[i];
 301                                 break;
 302                         }
 303                 }
 304                 nodemap2->num = 1;
 305
 306                 goto done;
 307         }
 308
 309         if (strcmp(nodestring, "all") == 0) {
 310                 for (i=0; i<nodemap->num; i++) {
 311                         nodemap2->node[i] = nodemap->node[i];
 312                 }
 313                 nodemap2->num = nodemap->num;
 314
 315                 goto done;
 316         } else {
 317                 char *ns, *tok;
 318
 319                 ns = talloc_strdup(mem_ctx, nodestring);
 320                 if (ns == NULL) {
 321                         return false;
 322                 }
 323
 324                 tok = strtok(ns, ",");
 325                 while (tok != NULL) {
 326                         uint32_t pnn;
 327                         char *endptr;
 328
 329                         pnn = (uint32_t)strtoul(tok, &endptr, 0);
 330                         if (pnn == 0 && tok == endptr) {
 331                                 fprintf(stderr, "Invalid node %s\n", tok);
 332                                         return false;
 333                         }
 334
 335                         node = get_node_by_pnn(nodemap, pnn);
 336                         if (node == NULL) {
 337                                 fprintf(stderr, "Node %u does not exist\n",
 338                                         pnn);
 339                                 return false;
 340                         }
 341
 342                         nodemap2->node[nodemap2->num] = *node;
 343                         nodemap2->num += 1;
 344
 345                         tok = strtok(NULL, ",");
 346                 }
 347         }
 348
 349 done:
 350         *out = nodemap2;
 351         return true;
 352 }
 353
 354 /* Compare IP address */
 355 static bool ctdb_same_ip(ctdb_sock_addr *ip1, ctdb_sock_addr *ip2)
 356 {
 357         bool ret = false;
 358
 359         if (ip1->sa.sa_family != ip2->sa.sa_family) {
 360                 return false;
 361         }
 362
 363         switch (ip1->sa.sa_family) {
 364         case AF_INET:
 365                 ret = (memcmp(&ip1->ip.sin_addr, &ip2->ip.sin_addr,
 366                               sizeof(struct in_addr)) == 0);
 367                 break;
 368
 369         case AF_INET6:
 370                 ret = (memcmp(&ip1->ip6.sin6_addr, &ip2->ip6.sin6_addr,
 371                               sizeof(struct in6_addr)) == 0);
 372                 break;
 373         }
 374
 375         return ret;
 376 }
 377
 378 /* Append a node to a node map with given address and flags */
 379 static bool node_map_add(struct ctdb_node_map *nodemap,
 380                          const char *nstr, uint32_t flags)
 381 {
 382         ctdb_sock_addr addr;
 383         uint32_t num;
 384         struct ctdb_node_and_flags *n;
 385
 386         if (! parse_ip(nstr, NULL, 0, &addr)) {
 387                 fprintf(stderr, "Invalid IP address %s\n", nstr);
 388                 return false;
 389         }
 390
 391         num = nodemap->num;
 392         nodemap->node = talloc_realloc(nodemap, nodemap->node,
 393                                        struct ctdb_node_and_flags, num+1);
 394         if (nodemap->node == NULL) {
 395                 return false;
 396         }
 397
 398         n = &nodemap->node[num];
 399         n->addr = addr;
 400         n->pnn = num;
 401         n->flags = flags;
 402
 403         nodemap->num = num+1;
 404         return true;
 405 }
 406
 407 /* Read a nodes file into a node map */
 408 static struct ctdb_node_map *ctdb_read_nodes_file(TALLOC_CTX *mem_ctx,
 409                                                   const char *nlist)
 410 {
 411         char **lines;
 412         int nlines;
 413         int i;
 414         struct ctdb_node_map *nodemap;
 415
 416         nodemap = talloc_zero(mem_ctx, struct ctdb_node_map);
 417         if (nodemap == NULL) {
 418                 return NULL;
 419         }
 420
 421         lines = file_lines_load(nlist, &nlines, 0, mem_ctx);
 422         if (lines == NULL) {
 423                 return NULL;
 424         }
 425
 426         while (nlines > 0 && strcmp(lines[nlines-1], "") == 0) {
 427                 nlines--;
 428         }
 429
 430         for (i=0; i<nlines; i++) {
 431                 char *node;
 432                 uint32_t flags;
 433                 size_t len;
 434
 435                 node = lines[i];
 436                 /* strip leading spaces */
 437                 while((*node == ' ') || (*node == '\t')) {
 438                         node++;
 439                 }
 440
 441                 len = strlen(node);
 442
 443                 /* strip trailing spaces */
 444                 while ((len > 1) &&
 445                        ((node[len-1] == ' ') || (node[len-1] == '\t')))
 446                 {
 447                         node[len-1] = '\0';
 448                         len--;
 449                 }
 450
 451                 if (len == 0) {
 452                         continue;
 453                 }
 454                 if (*node == '#') {
 455                         /* A "deleted" node is a node that is
 456                            commented out in the nodes file.  This is
 457                            used instead of removing a line, which
 458                            would cause subsequent nodes to change
 459                            their PNN. */
 460                         flags = NODE_FLAGS_DELETED;
 461                         node = discard_const("0.0.0.0");
 462                 } else {
 463                         flags = 0;
 464                 }
 465                 if (! node_map_add(nodemap, node, flags)) {
 466                         talloc_free(lines);
 467                         TALLOC_FREE(nodemap);
 468                         return NULL;
 469                 }
 470         }
 471
 472         talloc_free(lines);
 473         return nodemap;
 474 }
 475
 476 static struct ctdb_node_map *read_nodes_file(TALLOC_CTX *mem_ctx, uint32_t pnn)
 477 {
 478         struct ctdb_node_map *nodemap;
 479         char *nodepath;
 480         const char *nodes_list = NULL;
 481
 482         if (pnn != CTDB_UNKNOWN_PNN) {
 483                 nodepath = talloc_asprintf(mem_ctx, "CTDB_NODES_%u", pnn);
 484                 if (nodepath != NULL) {
 485                         nodes_list = getenv(nodepath);
 486                 }
 487         }
 488         if (nodes_list == NULL) {
 489                 nodes_list = getenv("CTDB_NODES");
 490         }
 491         if (nodes_list == NULL) {
 492                 const char *basedir = getenv("CTDB_BASE");
 493                 if (basedir == NULL) {
 494                         basedir = CTDB_ETCDIR;
 495                 }
 496                 nodes_list = talloc_asprintf(mem_ctx, "%s/nodes", basedir);
 497                 if (nodes_list == NULL) {
 498                         fprintf(stderr, "Memory allocation error\n");
 499                         return NULL;
 500                 }
 501         }
 502
 503         nodemap = ctdb_read_nodes_file(mem_ctx, nodes_list);
 504         if (nodemap == NULL) {
 505                 fprintf(stderr, "Failed to read nodes file \"%s\"\n",
 506                         nodes_list);
 507                 return NULL;
 508         }
 509
 510         return nodemap;
 511 }
 512
 513 static struct ctdb_dbid *db_find(TALLOC_CTX *mem_ctx,
 514                                  struct ctdb_context *ctdb,
 515                                  struct ctdb_dbid_map *dbmap,
 516                                  const char *db_name)
 517 {
 518         struct ctdb_dbid *db = NULL;
 519         const char *name;
 520         int ret, i;
 521
 522         for (i=0; i<dbmap->num; i++) {
 523                 ret = ctdb_ctrl_get_dbname(mem_ctx, ctdb->ev, ctdb->client,
 524                                            ctdb->pnn, TIMEOUT(),
 525                                            dbmap->dbs[i].db_id, &name);
 526                 if (ret != 0) {
 527                         return false;
 528                 }
 529
 530                 if (strcmp(db_name, name) == 0) {
 531                         talloc_free(discard_const(name));
 532                         db = &dbmap->dbs[i];
 533                         break;
 534                 }
 535         }
 536
 537         return db;
 538 }
 539
 540 static bool db_exists(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 541                       const char *db_arg, uint32_t *db_id,
 542                       const char **db_name, uint8_t *db_flags)
 543 {
 544         struct ctdb_dbid_map *dbmap;
 545         struct ctdb_dbid *db = NULL;
 546         uint32_t id = 0;
 547         const char *name = NULL;
 548         int ret, i;
 549
 550         ret = ctdb_ctrl_get_dbmap(mem_ctx, ctdb->ev, ctdb->client,
 551                                   ctdb->pnn, TIMEOUT(), &dbmap);
 552         if (ret != 0) {
 553                 return false;
 554         }
 555
 556         if (strncmp(db_arg, "0x", 2) == 0) {
 557                 id = strtoul(db_arg, NULL, 0);
 558                 for (i=0; i<dbmap->num; i++) {
 559                         if (id == dbmap->dbs[i].db_id) {
 560                                 db = &dbmap->dbs[i];
 561                                 break;
 562                         }
 563                 }
 564         } else {
 565                 name = db_arg;
 566                 db = db_find(mem_ctx, ctdb, dbmap, name);
 567         }
 568
 569         if (db == NULL) {
 570                 fprintf(stderr, "No database matching '%s' found\n", db_arg);
 571                 return false;
 572         }
 573
 574         if (name == NULL) {
 575                 ret = ctdb_ctrl_get_dbname(mem_ctx, ctdb->ev, ctdb->client,
 576                                            ctdb->pnn, TIMEOUT(), id, &name);
 577                 if (ret != 0) {
 578                         return false;
 579                 }
 580         }
 581
 582         if (db_id != NULL) {
 583                 *db_id = db->db_id;
 584         }
 585         if (db_name != NULL) {
 586                 *db_name = talloc_strdup(mem_ctx, name);
 587         }
 588         if (db_flags != NULL) {
 589                 *db_flags = db->flags;
 590         }
 591         return true;
 592 }
 593
 594 static int h2i(char h)
 595 {
 596         if (h >= 'a' && h <= 'f') {
 597                 return h - 'a' + 10;
 598         }
 599         if (h >= 'A' && h <= 'F') {
 600                 return h - 'f' + 10;
 601         }
 602         return h - '0';
 603 }
 604
 605 static int hex_to_data(const char *str, size_t len, TALLOC_CTX *mem_ctx,
 606                        TDB_DATA *out)
 607 {
 608         int i;
 609         TDB_DATA data;
 610
 611         if (len & 0x01) {
 612                 fprintf(stderr, "Key (%s) contains odd number of hex digits\n",
 613                         str);
 614                 return EINVAL;
 615         }
 616
 617         data.dsize = len / 2;
 618         data.dptr = talloc_size(mem_ctx, data.dsize);
 619         if (data.dptr == NULL) {
 620                 return ENOMEM;
 621         }
 622
 623         for (i=0; i<data.dsize; i++) {
 624                 data.dptr[i] = h2i(str[i*2]) << 4 | h2i(str[i*2+1]);
 625         }
 626
 627         *out = data;
 628         return 0;
 629 }
 630
 631 static int str_to_data(const char *str, size_t len, TALLOC_CTX *mem_ctx,
 632                        TDB_DATA *out)
 633 {
 634         TDB_DATA data;
 635         int ret = 0;
 636
 637         if (strncmp(str, "0x", 2) == 0) {
 638                 ret = hex_to_data(str+2, len-2, mem_ctx, &data);
 639         } else {
 640                 data.dptr = talloc_memdup(mem_ctx, str, len);
 641                 if (data.dptr == NULL) {
 642                         return ENOMEM;
 643                 }
 644                 data.dsize = len;
 645         }
 646
 647         *out = data;
 648         return ret;
 649 }
 650
 651 static int run_helper(TALLOC_CTX *mem_ctx, const char *command,
 652                       const char *path, int argc, const char **argv)
 653 {
 654         pid_t pid;
 655         int save_errno, status, ret;
 656         const char **new_argv;
 657         int i;
 658
 659         new_argv = talloc_array(mem_ctx, const char *, argc + 2);
 660         if (new_argv == NULL) {
 661                 return ENOMEM;
 662         }
 663
 664         new_argv[0] = path;
 665         for (i=0; i<argc; i++) {
 666                 new_argv[i+1] = argv[i];
 667         }
 668         new_argv[argc+1] = NULL;
 669
 670         pid = fork();
 671         if (pid < 0) {
 672                 save_errno = errno;
 673                 talloc_free(new_argv);
 674                 fprintf(stderr, "Failed to fork %s (%s) - %s\n",
 675                         command, path, strerror(save_errno));
 676                 return save_errno;
 677         }
 678
 679         if (pid == 0) {
 680                 ret = execv(path, discard_const(new_argv));
 681                 if (ret == -1) {
 682                         _exit(64+errno);
 683                 }
 684                 /* Should not happen */
 685                 _exit(64+ENOEXEC);
 686         }
 687
 688         talloc_free(new_argv);
 689
 690         ret = waitpid(pid, &status, 0);
 691         if (ret == -1) {
 692                 save_errno = errno;
 693                 fprintf(stderr, "waitpid() failed for %s - %s\n",
 694                         command, strerror(save_errno));
 695                 return save_errno;
 696         }
 697
 698         if (WIFEXITED(status)) {
 699                 int pstatus = WEXITSTATUS(status);
 700                 if (WIFSIGNALED(status)) {
 701                         fprintf(stderr, "%s terminated with signal %d\n",
 702                                 command, WTERMSIG(status));
 703                         ret = EINTR;
 704                 } else if (pstatus >= 64 && pstatus < 255) {
 705                         fprintf(stderr, "%s failed with error %d\n",
 706                                 command, pstatus-64);
 707                         ret = pstatus - 64;
 708                 } else {
 709                         ret = pstatus;
 710                 }
 711                 return ret;
 712         } else if (WIFSIGNALED(status)) {
 713                 fprintf(stderr, "%s terminated with signal %d\n",
 714                         command, WTERMSIG(status));
 715                 return EINTR;
 716         }
 717
 718         return 0;
 719 }
 720
 721 /*
 722  * Command Functions
 723  */
 724
 725 static int control_version(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 726                            int argc, const char **argv)
 727 {
 728         printf("%s\n", CTDB_VERSION_STRING);
 729         return 0;
 730 }
 731
 732 static bool partially_online(TALLOC_CTX *mem_ctx,
 733                              struct ctdb_context *ctdb,
 734                              struct ctdb_node_and_flags *node)
 735 {
 736         struct ctdb_iface_list *iface_list;
 737         int ret, i;
 738         bool status = false;
 739
 740         if (node->flags != 0) {
 741                 return false;
 742         }
 743
 744         ret = ctdb_ctrl_get_ifaces(mem_ctx, ctdb->ev, ctdb->client,
 745                                    node->pnn, TIMEOUT(), &iface_list);
 746         if (ret != 0) {
 747                 return false;
 748         }
 749
 750         status = false;
 751         for (i=0; i < iface_list->num; i++) {
 752                 if (iface_list->iface[i].link_state == 0) {
 753                         status = true;
 754                         break;
 755                 }
 756         }
 757
 758         return status;
 759 }
 760
 761 static void print_nodemap_machine(TALLOC_CTX *mem_ctx,
 762                                   struct ctdb_context *ctdb,
 763                                   struct ctdb_node_map *nodemap,
 764                                   uint32_t mypnn)
 765 {
 766         struct ctdb_node_and_flags *node;
 767         int i;
 768
 769         printf("%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s\n",
 770                options.sep,
 771                "Node", options.sep,
 772                "IP", options.sep,
 773                "Disconnected", options.sep,
 774                "Banned", options.sep,
 775                "Disabled", options.sep,
 776                "Unhealthy", options.sep,
 777                "Stopped", options.sep,
 778                "Inactive", options.sep,
 779                "PartiallyOnline", options.sep,
 780                "ThisNode", options.sep);
 781
 782         for (i=0; i<nodemap->num; i++) {
 783                 node = &nodemap->node[i];
 784                 if (node->flags & NODE_FLAGS_DELETED) {
 785                         continue;
 786                 }
 787
 788                 printf("%s%u%s%s%s%d%s%d%s%d%s%d%s%d%s%d%s%d%s%c%s\n",
 789                        options.sep,
 790                        node->pnn, options.sep,
 791                        ctdb_sock_addr_to_string(mem_ctx, &node->addr, false),
 792                        options.sep,
 793                        !! (node->flags & NODE_FLAGS_DISCONNECTED), options.sep,
 794                        !! (node->flags & NODE_FLAGS_BANNED), options.sep,
 795                        !! (node->flags & NODE_FLAGS_PERMANENTLY_DISABLED),
 796                        options.sep,
 797                        !! (node->flags & NODE_FLAGS_UNHEALTHY), options.sep,
 798                        !! (node->flags & NODE_FLAGS_STOPPED), options.sep,
 799                        !! (node->flags & NODE_FLAGS_INACTIVE), options.sep,
 800                        partially_online(mem_ctx, ctdb, node), options.sep,
 801                        (node->pnn == mypnn)?'Y':'N', options.sep);
 802         }
 803
 804 }
 805
 806 static void print_nodemap(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 807                           struct ctdb_node_map *nodemap, uint32_t mypnn,
 808                           bool print_header)
 809 {
 810         struct ctdb_node_and_flags *node;
 811         int num_deleted_nodes = 0;
 812         int i;
 813
 814         for (i=0; i<nodemap->num; i++) {
 815                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
 816                         num_deleted_nodes++;
 817                 }
 818         }
 819
 820         if (print_header) {
 821                 if (num_deleted_nodes == 0) {
 822                         printf("Number of nodes:%d\n", nodemap->num);
 823                 } else {
 824                         printf("Number of nodes:%d "
 825                                "(including %d deleted nodes)\n",
 826                                nodemap->num, num_deleted_nodes);
 827                 }
 828         }
 829
 830         for (i=0; i<nodemap->num; i++) {
 831                 node = &nodemap->node[i];
 832                 if (node->flags & NODE_FLAGS_DELETED) {
 833                         continue;
 834                 }
 835
 836                 printf("pnn:%u %-16s %s%s\n",
 837                        node->pnn,
 838                        ctdb_sock_addr_to_string(mem_ctx, &node->addr, false),
 839                        partially_online(mem_ctx, ctdb, node) ?
 840                                 "PARTIALLYONLINE" :
 841                                 pretty_print_flags(mem_ctx, node->flags),
 842                        node->pnn == mypnn ? " (THIS NODE)" : "");
 843         }
 844 }
 845
 846 static void print_status(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 847                          struct ctdb_node_map *nodemap, uint32_t mypnn,
 848                          struct ctdb_vnn_map *vnnmap, int recmode,
 849                          uint32_t recmaster)
 850 {
 851         int i;
 852
 853         print_nodemap(mem_ctx, ctdb, nodemap, mypnn, true);
 854
 855         if (vnnmap->generation == INVALID_GENERATION) {
 856                 printf("Generation:INVALID\n");
 857         } else {
 858                 printf("Generation:%u\n", vnnmap->generation);
 859         }
 860         printf("Size:%d\n", vnnmap->size);
 861         for (i=0; i<vnnmap->size; i++) {
 862                 printf("hash:%d lmaster:%d\n", i, vnnmap->map[i]);
 863         }
 864
 865         printf("Recovery mode:%s (%d)\n",
 866                recmode == CTDB_RECOVERY_NORMAL ? "NORMAL" : "RECOVERY",
 867                recmode);
 868         printf("Recovery master:%d\n", recmaster);
 869 }
 870
 871 static int control_status(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 872                           int argc, const char **argv)
 873 {
 874         struct ctdb_node_map *nodemap;
 875         struct ctdb_vnn_map *vnnmap;
 876         int recmode;
 877         uint32_t recmaster;
 878         int ret;
 879
 880         if (argc != 0) {
 881                 usage("status");
 882         }
 883
 884         nodemap = get_nodemap(ctdb, false);
 885         if (nodemap == NULL) {
 886                 return 1;
 887         }
 888
 889         if (options.machinereadable == 1) {
 890                 print_nodemap_machine(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn);
 891                 return 0;
 892         }
 893
 894         ret = ctdb_ctrl_getvnnmap(mem_ctx, ctdb->ev, ctdb->client,
 895                                   ctdb->cmd_pnn, TIMEOUT(), &vnnmap);
 896         if (ret != 0) {
 897                 return ret;
 898         }
 899
 900         ret = ctdb_ctrl_get_recmode(mem_ctx, ctdb->ev, ctdb->client,
 901                                     ctdb->cmd_pnn, TIMEOUT(), &recmode);
 902         if (ret != 0) {
 903                 return ret;
 904         }
 905
 906         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
 907                                       ctdb->cmd_pnn, TIMEOUT(), &recmaster);
 908         if (ret != 0) {
 909                 return ret;
 910         }
 911
 912         print_status(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn, vnnmap,
 913                      recmode, recmaster);
 914         return 0;
 915 }
 916
 917 static int control_uptime(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 918                           int argc, const char **argv)
 919 {
 920         struct ctdb_uptime *uptime;
 921         int ret, tmp, days, hours, minutes, seconds;
 922
 923         ret = ctdb_ctrl_uptime(mem_ctx, ctdb->ev, ctdb->client,
 924                                ctdb->cmd_pnn, TIMEOUT(), &uptime);
 925         if (ret != 0) {
 926                 return ret;
 927         }
 928
 929         printf("Current time of node %-4u     :                %s",
 930                ctdb->cmd_pnn, ctime(&uptime->current_time.tv_sec));
 931
 932         tmp = uptime->current_time.tv_sec - uptime->ctdbd_start_time.tv_sec;
 933         seconds = tmp % 60; tmp /= 60;
 934         minutes = tmp % 60; tmp /= 60;
 935         hours = tmp % 24; tmp /= 24;
 936         days = tmp;
 937
 938         printf("Ctdbd start time              : (%03d %02d:%02d:%02d) %s",
 939                days, hours, minutes, seconds,
 940                ctime(&uptime->ctdbd_start_time.tv_sec));
 941
 942         tmp = uptime->current_time.tv_sec - uptime->last_recovery_finished.tv_sec;
 943         seconds = tmp % 60; tmp /= 60;
 944         minutes = tmp % 60; tmp /= 60;
 945         hours = tmp % 24; tmp /= 24;
 946         days = tmp;
 947
 948         printf("Time of last recovery/failover: (%03d %02d:%02d:%02d) %s",
 949                days, hours, minutes, seconds,
 950                ctime(&uptime->last_recovery_finished.tv_sec));
 951
 952         printf("Duration of last recovery/failover: %lf seconds\n",
 953                timeval_delta(&uptime->last_recovery_finished,
 954                              &uptime->last_recovery_started));
 955
 956         return 0;
 957 }
 958
 959 static int control_ping(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 960                         int argc, const char **argv)
 961 {
 962         struct timeval tv;
 963         int ret, num_clients;
 964
 965         tv = timeval_current();
 966         ret = ctdb_ctrl_ping(mem_ctx, ctdb->ev, ctdb->client,
 967                              ctdb->cmd_pnn, TIMEOUT(), &num_clients);
 968         if (ret != 0) {
 969                 return ret;
 970         }
 971
 972         printf("response from %u time=%.6f sec  (%d clients)\n",
 973                ctdb->cmd_pnn, timeval_elapsed(&tv), num_clients);
 974         return 0;
 975 }
 976
 977 const char *runstate_to_string(enum ctdb_runstate runstate);
 978 enum ctdb_runstate runstate_from_string(const char *runstate_str);
 979
 980 static int control_runstate(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
 981                             int argc, const char **argv)
 982 {
 983         enum ctdb_runstate runstate;
 984         bool found;
 985         int ret, i;
 986
 987         ret = ctdb_ctrl_get_runstate(mem_ctx, ctdb->ev, ctdb->client,
 988                                      ctdb->cmd_pnn, TIMEOUT(), &runstate);
 989         if (ret != 0) {
 990                 return ret;
 991         }
 992
 993         found = true;
 994         for (i=0; i<argc; i++) {
 995                 enum ctdb_runstate t;
 996
 997                 found = false;
 998                 t = ctdb_runstate_from_string(argv[i]);
 999                 if (t == CTDB_RUNSTATE_UNKNOWN) {
1000                         printf("Invalid run state (%s)\n", argv[i]);
1001                         return 1;
1002                 }
1003
1004                 if (t == runstate) {
1005                         found = true;
1006                         break;
1007                 }
1008         }
1009
1010         if (! found) {
1011                 printf("CTDB not in required run state (got %s)\n",
1012                        ctdb_runstate_to_string(runstate));
1013                 return 1;
1014         }
1015
1016         printf("%s\n", ctdb_runstate_to_string(runstate));
1017         return 0;
1018 }
1019
1020 static int control_getvar(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1021                           int argc, const char **argv)
1022 {
1023         struct ctdb_var_list *tun_var_list;
1024         uint32_t value;
1025         int ret, i;
1026         bool found;
1027
1028         if (argc != 1) {
1029                 usage("getvar");
1030         }
1031
1032         ret = ctdb_ctrl_list_tunables(mem_ctx, ctdb->ev, ctdb->client,
1033                                       ctdb->cmd_pnn, TIMEOUT(), &tun_var_list);
1034         if (ret != 0) {
1035                 fprintf(stderr,
1036                         "Failed to get list of variables from node %u\n",
1037                         ctdb->cmd_pnn);
1038                 return ret;
1039         }
1040
1041         found = false;
1042         for (i=0; i<tun_var_list->count; i++) {
1043                 if (strcasecmp(tun_var_list->var[i], argv[0]) == 0) {
1044                         found = true;
1045                         break;
1046                 }
1047         }
1048
1049         if (! found) {
1050                 printf("No such tunable %s\n", argv[0]);
1051                 return 1;
1052         }
1053
1054         ret = ctdb_ctrl_get_tunable(mem_ctx, ctdb->ev, ctdb->client,
1055                                     ctdb->cmd_pnn, TIMEOUT(), argv[0], &value);
1056         if (ret != 0) {
1057                 return ret;
1058         }
1059
1060         printf("%-26s = %u\n", argv[0], value);
1061         return 0;
1062 }
1063
1064 static int control_setvar(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1065                           int argc, const char **argv)
1066 {
1067         struct ctdb_var_list *tun_var_list;
1068         struct ctdb_tunable tunable;
1069         int ret, i;
1070         bool found;
1071
1072         if (argc != 2) {
1073                 usage("setvar");
1074         }
1075
1076         ret = ctdb_ctrl_list_tunables(mem_ctx, ctdb->ev, ctdb->client,
1077                                       ctdb->cmd_pnn, TIMEOUT(), &tun_var_list);
1078         if (ret != 0) {
1079                 fprintf(stderr,
1080                         "Failed to get list of variables from node %u\n",
1081                         ctdb->cmd_pnn);
1082                 return ret;
1083         }
1084
1085         found = false;
1086         for (i=0; i<tun_var_list->count; i++) {
1087                 if (strcasecmp(tun_var_list->var[i], argv[0]) == 0) {
1088                         found = true;
1089                         break;
1090                 }
1091         }
1092
1093         if (! found) {
1094                 printf("No such tunable %s\n", argv[0]);
1095                 return 1;
1096         }
1097
1098         tunable.name = argv[0];
1099         tunable.value = strtoul(argv[1], NULL, 0);
1100
1101         ret = ctdb_ctrl_set_tunable(mem_ctx, ctdb->ev, ctdb->client,
1102                                     ctdb->cmd_pnn, TIMEOUT(), &tunable);
1103         if (ret != 0) {
1104                 if (ret == 1) {
1105                         fprintf(stderr,
1106                                 "Setting obsolete tunable variable '%s'\n",
1107                                tunable.name);
1108                         return 0;
1109                 }
1110         }
1111
1112         return ret;
1113 }
1114
1115 static int control_listvars(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1116                             int argc, const char **argv)
1117 {
1118         struct ctdb_var_list *tun_var_list;
1119         int ret, i;
1120
1121         if (argc != 0) {
1122                 usage("listvars");
1123         }
1124
1125         ret = ctdb_ctrl_list_tunables(mem_ctx, ctdb->ev, ctdb->client,
1126                                       ctdb->cmd_pnn, TIMEOUT(), &tun_var_list);
1127         if (ret != 0) {
1128                 return ret;
1129         }
1130
1131         for (i=0; i<tun_var_list->count; i++) {
1132                 control_getvar(mem_ctx, ctdb, 1, &tun_var_list->var[i]);
1133         }
1134
1135         return 0;
1136 }
1137
1138 const struct {
1139         const char *name;
1140         uint32_t offset;
1141 } stats_fields[] = {
1142 #define STATISTICS_FIELD(n) { #n, offsetof(struct ctdb_statistics, n) }
1143         STATISTICS_FIELD(num_clients),
1144         STATISTICS_FIELD(frozen),
1145         STATISTICS_FIELD(recovering),
1146         STATISTICS_FIELD(num_recoveries),
1147         STATISTICS_FIELD(client_packets_sent),
1148         STATISTICS_FIELD(client_packets_recv),
1149         STATISTICS_FIELD(node_packets_sent),
1150         STATISTICS_FIELD(node_packets_recv),
1151         STATISTICS_FIELD(keepalive_packets_sent),
1152         STATISTICS_FIELD(keepalive_packets_recv),
1153         STATISTICS_FIELD(node.req_call),
1154         STATISTICS_FIELD(node.reply_call),
1155         STATISTICS_FIELD(node.req_dmaster),
1156         STATISTICS_FIELD(node.reply_dmaster),
1157         STATISTICS_FIELD(node.reply_error),
1158         STATISTICS_FIELD(node.req_message),
1159         STATISTICS_FIELD(node.req_control),
1160         STATISTICS_FIELD(node.reply_control),
1161         STATISTICS_FIELD(client.req_call),
1162         STATISTICS_FIELD(client.req_message),
1163         STATISTICS_FIELD(client.req_control),
1164         STATISTICS_FIELD(timeouts.call),
1165         STATISTICS_FIELD(timeouts.control),
1166         STATISTICS_FIELD(timeouts.traverse),
1167         STATISTICS_FIELD(locks.num_calls),
1168         STATISTICS_FIELD(locks.num_current),
1169         STATISTICS_FIELD(locks.num_pending),
1170         STATISTICS_FIELD(locks.num_failed),
1171         STATISTICS_FIELD(total_calls),
1172         STATISTICS_FIELD(pending_calls),
1173         STATISTICS_FIELD(childwrite_calls),
1174         STATISTICS_FIELD(pending_childwrite_calls),
1175         STATISTICS_FIELD(memory_used),
1176         STATISTICS_FIELD(max_hop_count),
1177         STATISTICS_FIELD(total_ro_delegations),
1178         STATISTICS_FIELD(total_ro_revokes),
1179 };
1180
1181 #define LATENCY_AVG(v)  ((v).num ? (v).total / (v).num : 0.0 )
1182
1183 static void print_statistics_machine(struct ctdb_statistics *s,
1184                                      bool show_header)
1185 {
1186         int i;
1187
1188         if (show_header) {
1189                 printf("CTDB version%s", options.sep);
1190                 printf("Current time of statistics%s", options.sep);
1191                 printf("Statistics collected since%s", options.sep);
1192                 for (i=0; i<ARRAY_SIZE(stats_fields); i++) {
1193                         printf("%s%s", stats_fields[i].name, options.sep);
1194                 }
1195                 printf("num_reclock_ctdbd_latency%s", options.sep);
1196                 printf("min_reclock_ctdbd_latency%s", options.sep);
1197                 printf("avg_reclock_ctdbd_latency%s", options.sep);
1198                 printf("max_reclock_ctdbd_latency%s", options.sep);
1199
1200                 printf("num_reclock_recd_latency%s", options.sep);
1201                 printf("min_reclock_recd_latency%s", options.sep);
1202                 printf("avg_reclock_recd_latency%s", options.sep);
1203                 printf("max_reclock_recd_latency%s", options.sep);
1204
1205                 printf("num_call_latency%s", options.sep);
1206                 printf("min_call_latency%s", options.sep);
1207                 printf("avg_call_latency%s", options.sep);
1208                 printf("max_call_latency%s", options.sep);
1209
1210                 printf("num_lockwait_latency%s", options.sep);
1211                 printf("min_lockwait_latency%s", options.sep);
1212                 printf("avg_lockwait_latency%s", options.sep);
1213                 printf("max_lockwait_latency%s", options.sep);
1214
1215                 printf("num_childwrite_latency%s", options.sep);
1216                 printf("min_childwrite_latency%s", options.sep);
1217                 printf("avg_childwrite_latency%s", options.sep);
1218                 printf("max_childwrite_latency%s", options.sep);
1219                 printf("\n");
1220         }
1221
1222         printf("%u%s", CTDB_PROTOCOL, options.sep);
1223         printf("%u%s", (uint32_t)s->statistics_current_time.tv_sec, options.sep);
1224         printf("%u%s", (uint32_t)s->statistics_start_time.tv_sec, options.sep);
1225         for (i=0;i<ARRAY_SIZE(stats_fields);i++) {
1226                 printf("%u%s",
1227                        *(uint32_t *)(stats_fields[i].offset+(uint8_t *)s),
1228                        options.sep);
1229         }
1230         printf("%u%s", s->reclock.ctdbd.num, options.sep);
1231         printf("%.6f%s", s->reclock.ctdbd.min, options.sep);
1232         printf("%.6f%s", LATENCY_AVG(s->reclock.ctdbd), options.sep);
1233         printf("%.6f%s", s->reclock.ctdbd.max, options.sep);
1234
1235         printf("%u%s", s->reclock.recd.num, options.sep);
1236         printf("%.6f%s", s->reclock.recd.min, options.sep);
1237         printf("%.6f%s", LATENCY_AVG(s->reclock.recd), options.sep);
1238         printf("%.6f%s", s->reclock.recd.max, options.sep);
1239
1240         printf("%d%s", s->call_latency.num, options.sep);
1241         printf("%.6f%s", s->call_latency.min, options.sep);
1242         printf("%.6f%s", LATENCY_AVG(s->call_latency), options.sep);
1243         printf("%.6f%s", s->call_latency.max, options.sep);
1244
1245         printf("%d%s", s->childwrite_latency.num, options.sep);
1246         printf("%.6f%s", s->childwrite_latency.min, options.sep);
1247         printf("%.6f%s", LATENCY_AVG(s->childwrite_latency), options.sep);
1248         printf("%.6f%s", s->childwrite_latency.max, options.sep);
1249         printf("\n");
1250 }
1251
1252 static void print_statistics(struct ctdb_statistics *s)
1253 {
1254         int tmp, days, hours, minutes, seconds;
1255         int i;
1256         const char *prefix = NULL;
1257         int preflen = 0;
1258
1259         tmp = s->statistics_current_time.tv_sec -
1260               s->statistics_start_time.tv_sec;
1261         seconds = tmp % 60; tmp /= 60;
1262         minutes = tmp % 60; tmp /= 60;
1263         hours   = tmp % 24; tmp /= 24;
1264         days    = tmp;
1265
1266         printf("CTDB version %u\n", CTDB_PROTOCOL);
1267         printf("Current time of statistics  :                %s",
1268                ctime(&s->statistics_current_time.tv_sec));
1269         printf("Statistics collected since  : (%03d %02d:%02d:%02d) %s",
1270                days, hours, minutes, seconds,
1271                ctime(&s->statistics_start_time.tv_sec));
1272
1273         for (i=0; i<ARRAY_SIZE(stats_fields); i++) {
1274                 if (strchr(stats_fields[i].name, '.') != NULL) {
1275                         preflen = strcspn(stats_fields[i].name, ".") + 1;
1276                         if (! prefix ||
1277                             strncmp(prefix, stats_fields[i].name, preflen) != 0) {
1278                                 prefix = stats_fields[i].name;
1279                                 printf(" %*.*s\n", preflen-1, preflen-1,
1280                                        stats_fields[i].name);
1281                         }
1282                 } else {
1283                         preflen = 0;
1284                 }
1285                 printf(" %*s%-22s%*s%10u\n", preflen ? 4 : 0, "",
1286                        stats_fields[i].name+preflen, preflen ? 0 : 4, "",
1287                        *(uint32_t *)(stats_fields[i].offset+(uint8_t *)s));
1288         }
1289
1290         printf(" hop_count_buckets:");
1291         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
1292                 printf(" %d", s->hop_count_bucket[i]);
1293         }
1294         printf("\n");
1295         printf(" lock_buckets:");
1296         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
1297                 printf(" %d", s->locks.buckets[i]);
1298         }
1299         printf("\n");
1300         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
1301                "locks_latency      MIN/AVG/MAX",
1302                s->locks.latency.min, LATENCY_AVG(s->locks.latency),
1303                s->locks.latency.max, s->locks.latency.num);
1304
1305         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
1306                "reclock_ctdbd      MIN/AVG/MAX",
1307                s->reclock.ctdbd.min, LATENCY_AVG(s->reclock.ctdbd),
1308                s->reclock.ctdbd.max, s->reclock.ctdbd.num);
1309
1310         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
1311                "reclock_recd       MIN/AVG/MAX",
1312                s->reclock.recd.min, LATENCY_AVG(s->reclock.recd),
1313                s->reclock.recd.max, s->reclock.recd.num);
1314
1315         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
1316                "call_latency       MIN/AVG/MAX",
1317                s->call_latency.min, LATENCY_AVG(s->call_latency),
1318                s->call_latency.max, s->call_latency.num);
1319
1320         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
1321                "childwrite_latency MIN/AVG/MAX",
1322                s->childwrite_latency.min,
1323                LATENCY_AVG(s->childwrite_latency),
1324                s->childwrite_latency.max, s->childwrite_latency.num);
1325 }
1326
1327 static int control_statistics(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1328                               int argc, const char **argv)
1329 {
1330         struct ctdb_statistics *stats;
1331         int ret;
1332
1333         if (argc != 0) {
1334                 usage("statistics");
1335         }
1336
1337         ret = ctdb_ctrl_statistics(mem_ctx, ctdb->ev, ctdb->client,
1338                                    ctdb->cmd_pnn, TIMEOUT(), &stats);
1339         if (ret != 0) {
1340                 return ret;
1341         }
1342
1343         if (options.machinereadable) {
1344                 print_statistics_machine(stats, true);
1345         } else {
1346                 print_statistics(stats);
1347         }
1348
1349         return 0;
1350 }
1351
1352 static int control_statistics_reset(TALLOC_CTX *mem_ctx,
1353                                     struct ctdb_context *ctdb,
1354                                     int argc, const char **argv)
1355 {
1356         int ret;
1357
1358         if (argc != 0) {
1359                 usage("statisticsreset");
1360         }
1361
1362         ret = ctdb_ctrl_statistics_reset(mem_ctx, ctdb->ev, ctdb->client,
1363                                          ctdb->cmd_pnn, TIMEOUT());
1364         if (ret != 0) {
1365                 return ret;
1366         }
1367
1368         return 0;
1369 }
1370
1371 static int control_stats(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1372                          int argc, const char **argv)
1373 {
1374         struct ctdb_statistics_list *slist;
1375         int ret, count = 0, i;
1376         bool show_header = true;
1377
1378         if (argc > 1) {
1379                 usage("stats");
1380         }
1381
1382         if (argc == 1) {
1383                 count = atoi(argv[0]);
1384         }
1385
1386         ret = ctdb_ctrl_get_stat_history(mem_ctx, ctdb->ev, ctdb->client,
1387                                          ctdb->cmd_pnn, TIMEOUT(), &slist);
1388         if (ret != 0) {
1389                 return ret;
1390         }
1391
1392         for (i=0; i<slist->num; i++) {
1393                 if (slist->stats[i].statistics_start_time.tv_sec == 0) {
1394                         continue;
1395                 }
1396                 if (options.machinereadable == 1) {
1397                         print_statistics_machine(&slist->stats[i],
1398                                                  show_header);
1399                         show_header = false;
1400                 } else {
1401                         print_statistics(&slist->stats[i]);
1402                 }
1403                 if (count > 0 && i == count) {
1404                         break;
1405                 }
1406         }
1407
1408         return 0;
1409 }
1410
1411 static int ctdb_public_ip_cmp(const void *a, const void *b)
1412 {
1413         const struct ctdb_public_ip *ip_a = a;
1414         const struct ctdb_public_ip *ip_b = b;
1415
1416         return ctdb_sock_addr_cmp(&ip_a->addr, &ip_b->addr);
1417 }
1418
1419 static void print_ip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1420                      struct ctdb_public_ip_list *ips,
1421                      struct ctdb_public_ip_info **ipinfo,
1422                      bool all_nodes)
1423 {
1424         int i, j;
1425         char *conf, *avail, *active;
1426
1427         if (options.machinereadable == 1) {
1428                 printf("%s%s%s%s%s", options.sep,
1429                        "Public IP", options.sep,
1430                        "Node", options.sep);
1431                 if (options.verbose == 1) {
1432                         printf("%s%s%s%s%s%s\n",
1433                                "ActiveInterfaces", options.sep,
1434                                "AvailableInterfaces", options.sep,
1435                                "ConfiguredInterfaces", options.sep);
1436                 } else {
1437                         printf("\n");
1438                 }
1439         } else {
1440                 if (all_nodes) {
1441                         printf("Public IPs on ALL nodes\n");
1442                 } else {
1443                         printf("Public IPs on node %u\n", ctdb->cmd_pnn);
1444                 }
1445         }
1446
1447         for (i = 0; i < ips->num; i++) {
1448
1449                 if (options.machinereadable == 1) {
1450                         printf("%s%s%s%d%s", options.sep,
1451                                ctdb_sock_addr_to_string(
1452                                        mem_ctx, &ips->ip[i].addr, false),
1453                                options.sep,
1454                                (int)ips->ip[i].pnn, options.sep);
1455                 } else {
1456                         printf("%s", ctdb_sock_addr_to_string(
1457                                        mem_ctx, &ips->ip[i].addr, false));
1458                 }
1459
1460                 if (options.verbose == 0) {
1461                         if (options.machinereadable == 1) {
1462                                 printf("\n");
1463                         } else {
1464                                 printf(" %d\n", (int)ips->ip[i].pnn);
1465                         }
1466                         continue;
1467                 }
1468
1469                 conf = NULL;
1470                 avail = NULL;
1471                 active = NULL;
1472
1473                 if (ipinfo[i] == NULL) {
1474                         goto skip_ipinfo;
1475                 }
1476
1477                 for (j=0; j<ipinfo[i]->ifaces->num; j++) {
1478                         struct ctdb_iface *iface;
1479
1480                         iface = &ipinfo[i]->ifaces->iface[j];
1481                         if (conf == NULL) {
1482                                 conf = talloc_strdup(mem_ctx, iface->name);
1483                         } else {
1484                                 conf = talloc_asprintf_append(
1485                                                 conf, ",%s", iface->name);
1486                         }
1487
1488                         if (ipinfo[i]->active_idx == j) {
1489                                 active = iface->name;
1490                         }
1491
1492                         if (iface->link_state == 0) {
1493                                 continue;
1494                         }
1495
1496                         if (avail == NULL) {
1497                                 avail = talloc_strdup(mem_ctx, iface->name);
1498                         } else {
1499                                 avail = talloc_asprintf_append(
1500                                                 avail, ",%s", iface->name);
1501                         }
1502                 }
1503
1504         skip_ipinfo:
1505
1506                 if (options.machinereadable == 1) {
1507                         printf("%s%s%s%s%s%s\n",
1508                                active ? active : "", options.sep,
1509                                avail ? avail : "", options.sep,
1510                                conf ? conf : "", options.sep);
1511                 } else {
1512                         printf(" node[%d] active[%s] available[%s]"
1513                                " configured[%s]\n",
1514                                (int)ips->ip[i].pnn, active ? active : "",
1515                                avail ? avail : "", conf ? conf : "");
1516                 }
1517         }
1518 }
1519
1520 static int collect_ips(uint8_t *keybuf, size_t keylen, uint8_t *databuf,
1521                        size_t datalen, void *private_data)
1522 {
1523         struct ctdb_public_ip_list *ips = talloc_get_type_abort(
1524                 private_data, struct ctdb_public_ip_list);
1525         struct ctdb_public_ip *ip;
1526
1527         ip = (struct ctdb_public_ip *)databuf;
1528         ips->ip[ips->num] = *ip;
1529         ips->num += 1;
1530
1531         return 0;
1532 }
1533
1534 static int get_all_public_ips(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
1535                               struct ctdb_public_ip_list **out)
1536 {
1537         struct ctdb_node_map *nodemap;
1538         struct ctdb_public_ip_list *ips;
1539         struct db_hash_context *ipdb;
1540         uint32_t *pnn_list;
1541         int ret, count, i, j;
1542
1543         nodemap = get_nodemap(ctdb, false);
1544         if (nodemap == NULL) {
1545                 return 1;
1546         }
1547
1548         ret = db_hash_init(mem_ctx, "ips", 101, DB_HASH_COMPLEX, &ipdb);
1549         if (ret != 0) {
1550                 goto failed;
1551         }
1552
1553         count = list_of_active_nodes(nodemap, CTDB_UNKNOWN_PNN, mem_ctx,
1554                                      &pnn_list);
1555         if (count <= 0) {
1556                 goto failed;
1557         }
1558
1559         for (i=0; i<count; i++) {
1560                 ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
1561                                                pnn_list[i], TIMEOUT(),
1562                                                false, &ips);
1563                 if (ret != 0) {
1564                         goto failed;
1565                 }
1566
1567                 for (j=0; j<ips->num; j++) {
1568                         struct ctdb_public_ip ip;
1569
1570                         ip.pnn = ips->ip[j].pnn;
1571                         ip.addr = ips->ip[j].addr;
1572
1573                         if (pnn_list[i] == ip.pnn) {
1574                                 /* Node claims IP is hosted on it, so
1575                                  * save that information
1576                                  */
1577                                 ret = db_hash_add(ipdb, (uint8_t *)&ip.addr,
1578                                                   sizeof(ip.addr),
1579                                                   (uint8_t *)&ip, sizeof(ip));
1580                                 if (ret != 0) {
1581                                         goto failed;
1582                                 }
1583                         } else {
1584                                 /* Node thinks IP is hosted elsewhere,
1585                                  * so overwrite with CTDB_UNKNOWN_PNN
1586                                  * if there's no existing entry
1587                                  */
1588                                 ret = db_hash_exists(ipdb, (uint8_t *)&ip.addr,
1589                                                      sizeof(ip.addr));
1590                                 if (ret == ENOENT) {
1591                                         ip.pnn = CTDB_UNKNOWN_PNN;
1592                                         ret = db_hash_add(ipdb,
1593                                                           (uint8_t *)&ip.addr,
1594                                                           sizeof(ip.addr),
1595                                                           (uint8_t *)&ip,
1596                                                           sizeof(ip));
1597                                         if (ret != 0) {
1598                                                 goto failed;
1599                                         }
1600                                 }
1601                         }
1602                 }
1603
1604                 TALLOC_FREE(ips);
1605         }
1606
1607         talloc_free(pnn_list);
1608
1609         ret = db_hash_traverse(ipdb, NULL, NULL, &count);
1610         if (ret != 0) {
1611                 goto failed;
1612         }
1613
1614         ips = talloc_zero(mem_ctx, struct ctdb_public_ip_list);
1615         if (ips == NULL) {
1616                 goto failed;
1617         }
1618
1619         ips->ip = talloc_array(ips, struct ctdb_public_ip, count);
1620         if (ips->ip == NULL) {
1621                 goto failed;
1622         }
1623
1624         ret = db_hash_traverse(ipdb, collect_ips, ips, &count);
1625         if (ret != 0) {
1626                 goto failed;
1627         }
1628
1629         if (count != ips->num) {
1630                 goto failed;
1631         }
1632
1633         talloc_free(ipdb);
1634
1635         *out = ips;
1636         return 0;
1637
1638 failed:
1639         talloc_free(ipdb);
1640         return 1;
1641 }
1642
1643 static int control_ip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1644                       int argc, const char **argv)
1645 {
1646         struct ctdb_public_ip_list *ips;
1647         struct ctdb_public_ip_info **ipinfo;
1648         int ret, i;
1649         bool do_all = false;
1650
1651         if (argc > 1) {
1652                 usage("ip");
1653         }
1654
1655         if (argc == 1) {
1656                 if (strcmp(argv[0], "all") == 0) {
1657                         do_all = true;
1658                 } else {
1659                         usage("ip");
1660                 }
1661         }
1662
1663         if (do_all) {
1664                 ret = get_all_public_ips(ctdb, mem_ctx, &ips);
1665         } else {
1666                 ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
1667                                                ctdb->cmd_pnn, TIMEOUT(),
1668                                                false, &ips);
1669         }
1670         if (ret != 0) {
1671                 return ret;
1672         }
1673
1674         qsort(ips->ip, ips->num, sizeof(struct ctdb_public_ip),
1675               ctdb_public_ip_cmp);
1676
1677         ipinfo = talloc_array(mem_ctx, struct ctdb_public_ip_info *, ips->num);
1678         if (ipinfo == NULL) {
1679                 return 1;
1680         }
1681
1682         for (i=0; i<ips->num; i++) {
1683                 uint32_t pnn;
1684                 if (do_all) {
1685                         pnn = ips->ip[i].pnn;
1686                 } else {
1687                         pnn = ctdb->cmd_pnn;
1688                 }
1689                 if (pnn == CTDB_UNKNOWN_PNN) {
1690                         ipinfo[i] = NULL;
1691                         continue;
1692                 }
1693                 ret = ctdb_ctrl_get_public_ip_info(mem_ctx, ctdb->ev,
1694                                                    ctdb->client, pnn,
1695                                                    TIMEOUT(), &ips->ip[i].addr,
1696                                                    &ipinfo[i]);
1697                 if (ret != 0) {
1698                         return ret;
1699                 }
1700         }
1701
1702         print_ip(mem_ctx, ctdb, ips, ipinfo, do_all);
1703         return 0;
1704 }
1705
1706 static int control_ipinfo(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1707                           int argc, const char **argv)
1708 {
1709         struct ctdb_public_ip_info *ipinfo;
1710         ctdb_sock_addr addr;
1711         int ret, i;
1712
1713         if (argc != 1) {
1714                 usage("ipinfo");
1715         }
1716
1717         if (! parse_ip(argv[0], NULL, 0, &addr)) {
1718                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
1719                 return 1;
1720         }
1721
1722         ret = ctdb_ctrl_get_public_ip_info(mem_ctx, ctdb->ev, ctdb->client,
1723                                            ctdb->cmd_pnn, TIMEOUT(), &addr,
1724                                            &ipinfo);
1725         if (ret != 0) {
1726                 if (ret == -1) {
1727                         printf("Node %u does not know about IP %s\n",
1728                                ctdb->cmd_pnn, argv[0]);
1729                 }
1730                 return ret;
1731         }
1732
1733         printf("Public IP[%s] info on node %u\n",
1734                ctdb_sock_addr_to_string(mem_ctx, &ipinfo->ip.addr, false),
1735                                         ctdb->cmd_pnn);
1736
1737         printf("IP:%s\nCurrentNode:%u\nNumInterfaces:%u\n",
1738                ctdb_sock_addr_to_string(mem_ctx, &ipinfo->ip.addr, false),
1739                ipinfo->ip.pnn, ipinfo->ifaces->num);
1740
1741         for (i=0; i<ipinfo->ifaces->num; i++) {
1742                 struct ctdb_iface *iface;
1743
1744                 iface = &ipinfo->ifaces->iface[i];
1745                 iface->name[CTDB_IFACE_SIZE] = '\0';
1746                 printf("Interface[%u]: Name:%s Link:%s References:%u%s\n",
1747                        i+1, iface->name,
1748                        iface->link_state == 0 ? "down" : "up",
1749                        iface->references,
1750                        (i == ipinfo->active_idx) ? " (active)" : "");
1751         }
1752
1753         return 0;
1754 }
1755
1756 static int control_ifaces(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1757                           int argc, const char **argv)
1758 {
1759         struct ctdb_iface_list *ifaces;
1760         int ret, i;
1761
1762         if (argc != 0) {
1763                 usage("ifaces");
1764         }
1765
1766         ret = ctdb_ctrl_get_ifaces(mem_ctx, ctdb->ev, ctdb->client,
1767                                    ctdb->cmd_pnn, TIMEOUT(), &ifaces);
1768         if (ret != 0) {
1769                 return ret;
1770         }
1771
1772         if (ifaces->num == 0) {
1773                 printf("No interfaces configured on node %u\n",
1774                        ctdb->cmd_pnn);
1775                 return 0;
1776         }
1777
1778         if (options.machinereadable) {
1779                 printf("%s%s%s%s%s%s%s\n", options.sep,
1780                        "Name", options.sep,
1781                        "LinkStatus", options.sep,
1782                        "References", options.sep);
1783         } else {
1784                 printf("Interfaces on node %u\n", ctdb->cmd_pnn);
1785         }
1786
1787         for (i=0; i<ifaces->num; i++) {
1788                 if (options.machinereadable) {
1789                         printf("%s%s%s%u%s%u%s\n", options.sep,
1790                                ifaces->iface[i].name, options.sep,
1791                                ifaces->iface[i].link_state, options.sep,
1792                                ifaces->iface[i].references, options.sep);
1793                 } else {
1794                         printf("name:%s link:%s references:%u\n",
1795                                ifaces->iface[i].name,
1796                                ifaces->iface[i].link_state ? "up" : "down",
1797                                ifaces->iface[i].references);
1798                 }
1799         }
1800
1801         return 0;
1802 }
1803
1804 static int control_setifacelink(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1805                                 int argc, const char **argv)
1806 {
1807         struct ctdb_iface_list *ifaces;
1808         struct ctdb_iface *iface;
1809         int ret, i;
1810
1811         if (argc != 2) {
1812                 usage("setifacelink");
1813         }
1814
1815         if (strlen(argv[0]) > CTDB_IFACE_SIZE) {
1816                 fprintf(stderr, "Interface name '%s' too long\n", argv[0]);
1817                 return 1;
1818         }
1819
1820         ret = ctdb_ctrl_get_ifaces(mem_ctx, ctdb->ev, ctdb->client,
1821                                    ctdb->cmd_pnn, TIMEOUT(), &ifaces);
1822         if (ret != 0) {
1823                 fprintf(stderr,
1824                         "Failed to get interface information from node %u\n",
1825                         ctdb->cmd_pnn);
1826                 return ret;
1827         }
1828
1829         iface = NULL;
1830         for (i=0; i<ifaces->num; i++) {
1831                 if (strcmp(ifaces->iface[i].name, argv[0]) == 0) {
1832                         iface = &ifaces->iface[i];
1833                         break;
1834                 }
1835         }
1836
1837         if (iface == NULL) {
1838                 printf("Interface %s not configured on node %u\n",
1839                        argv[0], ctdb->cmd_pnn);
1840                 return 1;
1841         }
1842
1843         if (strcmp(argv[1], "up") == 0) {
1844                 iface->link_state = 1;
1845         } else if (strcmp(argv[1], "down") == 0) {
1846                 iface->link_state = 0;
1847         } else {
1848                 usage("setifacelink");
1849                 return 1;
1850         }
1851
1852         iface->references = 0;
1853
1854         ret = ctdb_ctrl_set_iface_link_state(mem_ctx, ctdb->ev, ctdb->client,
1855                                              ctdb->cmd_pnn, TIMEOUT(), iface);
1856         if (ret != 0) {
1857                 return ret;
1858         }
1859
1860         return 0;
1861 }
1862
1863 static int control_process_exists(TALLOC_CTX *mem_ctx,
1864                                   struct ctdb_context *ctdb,
1865                                   int argc, const char **argv)
1866 {
1867         pid_t pid;
1868         uint64_t srvid = 0;
1869         int ret, status;
1870
1871         if (argc != 1 && argc != 2) {
1872                 usage("process-exists");
1873         }
1874
1875         pid = atoi(argv[0]);
1876         if (argc == 2) {
1877                 srvid = strtoull(argv[1], NULL, 0);
1878         }
1879
1880         if (srvid == 0) {
1881                 ret = ctdb_ctrl_process_exists(mem_ctx, ctdb->ev, ctdb->client,
1882                                        ctdb->cmd_pnn, TIMEOUT(), pid, &status);
1883         } else {
1884                 struct ctdb_pid_srvid pid_srvid;
1885
1886                 pid_srvid.pid = pid;
1887                 pid_srvid.srvid = srvid;
1888
1889                 ret = ctdb_ctrl_check_pid_srvid(mem_ctx, ctdb->ev,
1890                                                 ctdb->client, ctdb->cmd_pnn,
1891                                                 TIMEOUT(), &pid_srvid,
1892                                                 &status);
1893         }
1894
1895         if (ret != 0) {
1896                 return ret;
1897         }
1898
1899         if (srvid == 0) {
1900                 printf("PID %d %s\n", pid,
1901                        (status == 0 ? "exists" : "does not exist"));
1902         } else {
1903                 printf("PID %d with SRVID 0x%"PRIx64" %s\n", pid, srvid,
1904                        (status == 0 ? "exists" : "does not exist"));
1905         }
1906         return status;
1907 }
1908
1909 static int control_getdbmap(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1910                             int argc, const char **argv)
1911 {
1912         struct ctdb_dbid_map *dbmap;
1913         int ret, i;
1914
1915         if (argc != 0) {
1916                 usage("getdbmap");
1917         }
1918
1919         ret = ctdb_ctrl_get_dbmap(mem_ctx, ctdb->ev, ctdb->client,
1920                                   ctdb->cmd_pnn, TIMEOUT(), &dbmap);
1921         if (ret != 0) {
1922                 return ret;
1923         }
1924
1925         if (options.machinereadable == 1) {
1926                 printf("%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s\n",
1927                        options.sep,
1928                        "ID", options.sep,
1929                        "Name", options.sep,
1930                        "Path", options.sep,
1931                        "Persistent", options.sep,
1932                        "Sticky", options.sep,
1933                        "Unhealthy", options.sep,
1934                        "Readonly", options.sep,
1935                        "Replicated", options.sep);
1936         } else {
1937                 printf("Number of databases:%d\n", dbmap->num);
1938         }
1939
1940         for (i=0; i<dbmap->num; i++) {
1941                 const char *name;
1942                 const char *path;
1943                 const char *health;
1944                 bool persistent;
1945                 bool readonly;
1946                 bool sticky;
1947                 bool replicated;
1948                 uint32_t db_id;
1949
1950                 db_id = dbmap->dbs[i].db_id;
1951
1952                 ret = ctdb_ctrl_get_dbname(mem_ctx, ctdb->ev, ctdb->client,
1953                                            ctdb->cmd_pnn, TIMEOUT(), db_id,
1954                                            &name);
1955                 if (ret != 0) {
1956                         return ret;
1957                 }
1958
1959                 ret = ctdb_ctrl_getdbpath(mem_ctx, ctdb->ev, ctdb->client,
1960                                           ctdb->cmd_pnn, TIMEOUT(), db_id,
1961                                           &path);
1962                 if (ret != 0) {
1963                         return ret;
1964                 }
1965
1966                 ret = ctdb_ctrl_db_get_health(mem_ctx, ctdb->ev, ctdb->client,
1967                                               ctdb->cmd_pnn, TIMEOUT(), db_id,
1968                                               &health);
1969                 if (ret != 0) {
1970                         return ret;
1971                 }
1972
1973                 persistent = dbmap->dbs[i].flags & CTDB_DB_FLAGS_PERSISTENT;
1974                 readonly = dbmap->dbs[i].flags & CTDB_DB_FLAGS_READONLY;
1975                 sticky = dbmap->dbs[i].flags & CTDB_DB_FLAGS_STICKY;
1976                 replicated = dbmap->dbs[i].flags & CTDB_DB_FLAGS_REPLICATED;
1977
1978                 if (options.machinereadable == 1) {
1979                         printf("%s0x%08X%s%s%s%s%s%d%s%d%s%d%s%d%s%d%s\n",
1980                                options.sep,
1981                                db_id, options.sep,
1982                                name, options.sep,
1983                                path, options.sep,
1984                                !! (persistent), options.sep,
1985                                !! (sticky), options.sep,
1986                                !! (health), options.sep,
1987                                !! (readonly), options.sep,
1988                                !! (replicated), options.sep);
1989                 } else {
1990                         printf("dbid:0x%08x name:%s path:%s%s%s%s%s%s\n",
1991                                db_id, name, path,
1992                                persistent ? " PERSISTENT" : "",
1993                                sticky ? " STICKY" : "",
1994                                readonly ? " READONLY" : "",
1995                                replicated ? " REPLICATED" : "",
1996                                health ? " UNHEALTHY" : "");
1997                 }
1998
1999                 talloc_free(discard_const(name));
2000                 talloc_free(discard_const(path));
2001                 talloc_free(discard_const(health));
2002         }
2003
2004         return 0;
2005 }
2006
2007 static int control_getdbstatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2008                                int argc, const char **argv)
2009 {
2010         uint32_t db_id;
2011         const char *db_name, *db_path, *db_health;
2012         uint8_t db_flags;
2013         int ret;
2014
2015         if (argc != 1) {
2016                 usage("getdbstatus");
2017         }
2018
2019         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
2020                 return 1;
2021         }
2022
2023         ret = ctdb_ctrl_getdbpath(mem_ctx, ctdb->ev, ctdb->client,
2024                                   ctdb->cmd_pnn, TIMEOUT(), db_id,
2025                                   &db_path);
2026         if (ret != 0) {
2027                 return ret;
2028         }
2029
2030         ret = ctdb_ctrl_db_get_health(mem_ctx, ctdb->ev, ctdb->client,
2031                                       ctdb->cmd_pnn, TIMEOUT(), db_id,
2032                                       &db_health);
2033         if (ret != 0) {
2034                 return ret;
2035         }
2036
2037         printf("dbid: 0x%08x\nname: %s\npath: %s\n", db_id, db_name, db_path);
2038         printf("PERSISTENT: %s\nREPLICATED: %s\nSTICKY: %s\nREADONLY: %s\n",
2039                (db_flags & CTDB_DB_FLAGS_PERSISTENT ? "yes" : "no"),
2040                (db_flags & CTDB_DB_FLAGS_REPLICATED ? "yes" : "no"),
2041                (db_flags & CTDB_DB_FLAGS_STICKY ? "yes" : "no"),
2042                (db_flags & CTDB_DB_FLAGS_READONLY ? "yes" : "no"));
2043         printf("HEALTH: %s\n", (db_health ? db_health : "OK"));
2044         return 0;
2045 }
2046
2047 struct dump_record_state {
2048         uint32_t count;
2049 };
2050
2051 #define ISASCII(x) (isprint(x) && ! strchr("\"\\", (x)))
2052
2053 static void dump_tdb_data(const char *name, TDB_DATA val)
2054 {
2055         int i;
2056
2057         fprintf(stdout, "%s(%zu) = \"", name, val.dsize);
2058         for (i=0; i<val.dsize; i++) {
2059                 if (ISASCII(val.dptr[i])) {
2060                         fprintf(stdout, "%c", val.dptr[i]);
2061                 } else {
2062                         fprintf(stdout, "\\%02X", val.dptr[i]);
2063                 }
2064         }
2065         fprintf(stdout, "\"\n");
2066 }
2067
2068 static void dump_ltdb_header(struct ctdb_ltdb_header *header)
2069 {
2070         fprintf(stdout, "dmaster: %u\n", header->dmaster);
2071         fprintf(stdout, "rsn: %" PRIu64 "\n", header->rsn);
2072         fprintf(stdout, "flags: 0x%08x", header->flags);
2073         if (header->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) {
2074                 fprintf(stdout, " MIGRATED_WITH_DATA");
2075         }
2076         if (header->flags & CTDB_REC_FLAG_VACUUM_MIGRATED) {
2077                 fprintf(stdout, " VACUUM_MIGRATED");
2078         }
2079         if (header->flags & CTDB_REC_FLAG_AUTOMATIC) {
2080                 fprintf(stdout, " AUTOMATIC");
2081         }
2082         if (header->flags & CTDB_REC_RO_HAVE_DELEGATIONS) {
2083                 fprintf(stdout, " RO_HAVE_DELEGATIONS");
2084         }
2085         if (header->flags & CTDB_REC_RO_HAVE_READONLY) {
2086                 fprintf(stdout, " RO_HAVE_READONLY");
2087         }
2088         if (header->flags & CTDB_REC_RO_REVOKING_READONLY) {
2089                 fprintf(stdout, " RO_REVOKING_READONLY");
2090         }
2091         if (header->flags & CTDB_REC_RO_REVOKE_COMPLETE) {
2092                 fprintf(stdout, " RO_REVOKE_COMPLETE");
2093         }
2094         fprintf(stdout, "\n");
2095
2096 }
2097
2098 static int dump_record(uint32_t reqid, struct ctdb_ltdb_header *header,
2099                        TDB_DATA key, TDB_DATA data, void *private_data)
2100 {
2101         struct dump_record_state *state =
2102                 (struct dump_record_state *)private_data;
2103
2104         state->count += 1;
2105
2106         dump_tdb_data("key", key);
2107         dump_ltdb_header(header);
2108         dump_tdb_data("data", data);
2109         fprintf(stdout, "\n");
2110
2111         return 0;
2112 }
2113
2114 static int control_catdb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2115                          int argc, const char **argv)
2116 {
2117         struct ctdb_db_context *db;
2118         const char *db_name;
2119         uint32_t db_id;
2120         uint8_t db_flags;
2121         struct dump_record_state state;
2122         int ret;
2123
2124         if (argc != 1) {
2125                 usage("catdb");
2126         }
2127
2128         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
2129                 return 1;
2130         }
2131
2132         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
2133                           db_flags, &db);
2134         if (ret != 0) {
2135                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
2136                 return ret;
2137         }
2138
2139         state.count = 0;
2140
2141         ret = ctdb_db_traverse(mem_ctx, ctdb->ev, ctdb->client, db,
2142                                ctdb->cmd_pnn, TIMEOUT(),
2143                                dump_record, &state);
2144
2145         printf("Dumped %u records\n", state.count);
2146
2147         return ret;
2148 }
2149
2150 static int control_cattdb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2151                           int argc, const char **argv)
2152 {
2153         struct ctdb_db_context *db;
2154         const char *db_name;
2155         uint32_t db_id;
2156         uint8_t db_flags;
2157         struct dump_record_state state;
2158         int ret;
2159
2160         if (argc != 1) {
2161                 usage("catdb");
2162         }
2163
2164         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
2165                 return 1;
2166         }
2167
2168         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
2169                           db_flags, &db);
2170         if (ret != 0) {
2171                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
2172                 return ret;
2173         }
2174
2175         state.count = 0;
2176         ret = ctdb_db_traverse_local(db, true, true, dump_record, &state);
2177
2178         printf("Dumped %u record(s)\n", state.count);
2179
2180         return ret;
2181 }
2182
2183 static int control_getcapabilities(TALLOC_CTX *mem_ctx,
2184                                    struct ctdb_context *ctdb,
2185                                    int argc, const char **argv)
2186 {
2187         uint32_t caps;
2188         int ret;
2189
2190         if (argc != 0) {
2191                 usage("getcapabilities");
2192         }
2193
2194         ret = ctdb_ctrl_get_capabilities(mem_ctx, ctdb->ev, ctdb->client,
2195                                          ctdb->cmd_pnn, TIMEOUT(), &caps);
2196         if (ret != 0) {
2197                 return ret;
2198         }
2199
2200         if (options.machinereadable == 1) {
2201                 printf("%s%s%s%s%s\n",
2202                        options.sep,
2203                        "RECMASTER", options.sep,
2204                        "LMASTER", options.sep);
2205                 printf("%s%d%s%d%s\n", options.sep,
2206                        !! (caps & CTDB_CAP_RECMASTER), options.sep,
2207                        !! (caps & CTDB_CAP_LMASTER), options.sep);
2208         } else {
2209                 printf("RECMASTER: %s\n",
2210                        (caps & CTDB_CAP_RECMASTER) ? "YES" : "NO");
2211                 printf("LMASTER: %s\n",
2212                        (caps & CTDB_CAP_LMASTER) ? "YES" : "NO");
2213         }
2214
2215         return 0;
2216 }
2217
2218 static int control_pnn(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2219                        int argc, const char **argv)
2220 {
2221         printf("%u\n", ctdb_client_pnn(ctdb->client));
2222         return 0;
2223 }
2224
2225 static int control_lvs(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2226                        int argc, const char **argv)
2227 {
2228         char *t, *lvs_helper = NULL;
2229
2230         if (argc != 1) {
2231                 usage("lvs");
2232         }
2233
2234         t = getenv("CTDB_LVS_HELPER");
2235         if (t != NULL) {
2236                 lvs_helper = talloc_strdup(mem_ctx, t);
2237         } else {
2238                 lvs_helper = talloc_asprintf(mem_ctx, "%s/ctdb_lvs",
2239                                              CTDB_HELPER_BINDIR);
2240         }
2241
2242         if (lvs_helper == NULL) {
2243                 fprintf(stderr, "Unable to set LVS helper\n");
2244                 return 1;
2245         }
2246
2247         return run_helper(mem_ctx, "LVS helper", lvs_helper, argc, argv);
2248 }
2249
2250 static int control_setdebug(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2251                             int argc, const char **argv)
2252 {
2253         int log_level;
2254         int ret;
2255         bool found;
2256
2257         if (argc != 1) {
2258                 usage("setdebug");
2259         }
2260
2261         found = debug_level_parse(argv[0], &log_level);
2262         if (! found) {
2263                 fprintf(stderr,
2264                         "Invalid debug level '%s'. Valid levels are:\n",
2265                         argv[0]);
2266                 fprintf(stderr, "\tERROR | WARNING | NOTICE | INFO | DEBUG\n");
2267                 return 1;
2268         }
2269
2270         ret = ctdb_ctrl_setdebug(mem_ctx, ctdb->ev, ctdb->client,
2271                                  ctdb->cmd_pnn, TIMEOUT(), log_level);
2272         if (ret != 0) {
2273                 return ret;
2274         }
2275
2276         return 0;
2277 }
2278
2279 static int control_getdebug(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2280                             int argc, const char **argv)
2281 {
2282         int loglevel;
2283         const char *log_str;
2284         int ret;
2285
2286         if (argc != 0) {
2287                 usage("getdebug");
2288         }
2289
2290         ret = ctdb_ctrl_getdebug(mem_ctx, ctdb->ev, ctdb->client,
2291                                  ctdb->cmd_pnn, TIMEOUT(), &loglevel);
2292         if (ret != 0) {
2293                 return ret;
2294         }
2295
2296         log_str = debug_level_to_string(loglevel);
2297         printf("%s\n", log_str);
2298
2299         return 0;
2300 }
2301
2302 static int control_attach(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2303                           int argc, const char **argv)
2304 {
2305         const char *db_name;
2306         uint8_t db_flags = 0;
2307         int ret;
2308
2309         if (argc < 1 || argc > 2) {
2310                 usage("attach");
2311         }
2312
2313         db_name = argv[0];
2314         if (argc == 2) {
2315                 if (strcmp(argv[1], "persistent") == 0) {
2316                         db_flags = CTDB_DB_FLAGS_PERSISTENT;
2317                 } else if (strcmp(argv[1], "readonly") == 0) {
2318                         db_flags = CTDB_DB_FLAGS_READONLY;
2319                 } else if (strcmp(argv[1], "sticky") == 0) {
2320                         db_flags = CTDB_DB_FLAGS_STICKY;
2321                 } else if (strcmp(argv[1], "replicated") == 0) {
2322                         db_flags = CTDB_DB_FLAGS_REPLICATED;
2323                 } else {
2324                         usage("attach");
2325                 }
2326         }
2327
2328         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
2329                           db_flags, NULL);
2330         if (ret != 0) {
2331                 return ret;
2332         }
2333
2334         return 0;
2335 }
2336
2337 static int control_detach(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2338                           int argc, const char **argv)
2339 {
2340         const char *db_name;
2341         uint32_t db_id;
2342         uint8_t db_flags;
2343         struct ctdb_node_map *nodemap;
2344         int recmode;
2345         int ret, ret2, i;
2346
2347         if (argc < 1) {
2348                 usage("detach");
2349         }
2350
2351         ret = ctdb_ctrl_get_recmode(mem_ctx, ctdb->ev, ctdb->client,
2352                                     ctdb->cmd_pnn, TIMEOUT(), &recmode);
2353         if (ret != 0) {
2354                 return ret;
2355         }
2356
2357         if (recmode == CTDB_RECOVERY_ACTIVE) {
2358                 fprintf(stderr, "Database cannot be detached"
2359                                 " when recovery is active\n");
2360                 return 1;
2361         }
2362
2363         nodemap = get_nodemap(ctdb, false);
2364         if (nodemap == NULL) {
2365                 return 1;
2366         }
2367
2368         for (i=0; i<nodemap->num; i++) {
2369                 uint32_t value;
2370
2371                 if (nodemap->node[i].flags & NODE_FLAGS_DISCONNECTED) {
2372                         continue;
2373                 }
2374                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
2375                         continue;
2376                 }
2377                 if (nodemap->node[i].flags & NODE_FLAGS_INACTIVE) {
2378                         fprintf(stderr, "Database cannot be detached on"
2379                                 " inactive (stopped or banned) node %u\n",
2380                                 nodemap->node[i].pnn);
2381                         return 1;
2382                 }
2383
2384                 ret = ctdb_ctrl_get_tunable(mem_ctx, ctdb->ev, ctdb->client,
2385                                             nodemap->node[i].pnn, TIMEOUT(),
2386                                             "AllowClientDBAttach", &value);
2387                 if (ret != 0) {
2388                         fprintf(stderr,
2389                                 "Unable to get tunable AllowClientDBAttach"
2390                                 " from node %u\n", nodemap->node[i].pnn);
2391                         return ret;
2392                 }
2393
2394                 if (value == 1) {
2395                         fprintf(stderr,
2396                                 "Database access is still active on node %u."
2397                                 " Set AllowclientDBAttach=0 on all nodes.\n",
2398                                 nodemap->node[i].pnn);
2399                         return 1;
2400                 }
2401         }
2402
2403         ret2 = 0;
2404         for (i=0; i<argc; i++) {
2405                 if (! db_exists(mem_ctx, ctdb, argv[i], &db_id, &db_name,
2406                                 &db_flags)) {
2407                         continue;
2408                 }
2409
2410                 if (db_flags &
2411                     (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
2412                         fprintf(stderr,
2413                                 "Only volatile databases can be detached\n");
2414                         return 1;
2415                 }
2416
2417                 ret = ctdb_detach(ctdb->ev, ctdb->client, TIMEOUT(), db_id);
2418                 if (ret != 0) {
2419                         fprintf(stderr, "Database %s detach failed\n", db_name);
2420                         ret2 = ret;
2421                 }
2422         }
2423
2424         return ret2;
2425 }
2426
2427 static int control_dumpmemory(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2428                               int argc, const char **argv)
2429 {
2430         const char *mem_str;
2431         ssize_t n;
2432         int ret;
2433
2434         ret = ctdb_ctrl_dump_memory(mem_ctx, ctdb->ev, ctdb->client,
2435                                     ctdb->cmd_pnn, TIMEOUT(), &mem_str);
2436         if (ret != 0) {
2437                 return ret;
2438         }
2439
2440         n = write(1, mem_str, strlen(mem_str)+1);
2441         if (n < 0 || n != strlen(mem_str)+1) {
2442                 fprintf(stderr, "Failed to write talloc summary\n");
2443                 return 1;
2444         }
2445
2446         return 0;
2447 }
2448
2449 static void dump_memory(uint64_t srvid, TDB_DATA data, void *private_data)
2450 {
2451         bool *done = (bool *)private_data;
2452         ssize_t n;
2453
2454         n = write(1, data.dptr, data.dsize);
2455         if (n < 0 || n != data.dsize) {
2456                 fprintf(stderr, "Failed to write talloc summary\n");
2457         }
2458
2459         *done = true;
2460 }
2461
2462 static int control_rddumpmemory(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2463                                 int argc, const char **argv)
2464 {
2465         struct ctdb_srvid_message msg = { 0 };
2466         int ret;
2467         bool done = false;
2468
2469         msg.pnn = ctdb->pnn;
2470         msg.srvid = next_srvid(ctdb);
2471
2472         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
2473                                               msg.srvid, dump_memory, &done);
2474         if (ret != 0) {
2475                 return ret;
2476         }
2477
2478         ret = ctdb_message_mem_dump(mem_ctx, ctdb->ev, ctdb->client,
2479                                     ctdb->cmd_pnn, &msg);
2480         if (ret != 0) {
2481                 return ret;
2482         }
2483
2484         ctdb_client_wait(ctdb->ev, &done);
2485         return 0;
2486 }
2487
2488 static int control_getpid(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2489                           int argc, const char **argv)
2490 {
2491         pid_t pid;
2492         int ret;
2493
2494         ret = ctdb_ctrl_get_pid(mem_ctx, ctdb->ev, ctdb->client,
2495                                 ctdb->cmd_pnn, TIMEOUT(), &pid);
2496         if (ret != 0) {
2497                 return ret;
2498         }
2499
2500         printf("%u\n", pid);
2501         return 0;
2502 }
2503
2504 static int check_flags(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2505                        const char *desc, uint32_t flag, bool set_flag)
2506 {
2507         struct ctdb_node_map *nodemap;
2508         bool flag_is_set;
2509
2510         nodemap = get_nodemap(ctdb, false);
2511         if (nodemap == NULL) {
2512                 return 1;
2513         }
2514
2515         flag_is_set = nodemap->node[ctdb->cmd_pnn].flags & flag;
2516         if (set_flag == flag_is_set) {
2517                 if (set_flag) {
2518                         fprintf(stderr, "Node %u is already %s\n",
2519                                 ctdb->cmd_pnn, desc);
2520                 } else {
2521                         fprintf(stderr, "Node %u is not %s\n",
2522                                 ctdb->cmd_pnn, desc);
2523                 }
2524                 return 0;
2525         }
2526
2527         return 1;
2528 }
2529
2530 static void wait_for_flags(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2531                            uint32_t flag, bool set_flag)
2532 {
2533         struct ctdb_node_map *nodemap;
2534         bool flag_is_set;
2535
2536         while (1) {
2537                 nodemap = get_nodemap(ctdb, true);
2538                 if (nodemap == NULL) {
2539                         fprintf(stderr,
2540                                 "Failed to get nodemap, trying again\n");
2541                         sleep(1);
2542                         continue;
2543                 }
2544
2545                 flag_is_set = nodemap->node[ctdb->cmd_pnn].flags & flag;
2546                 if (flag_is_set == set_flag) {
2547                         break;
2548                 }
2549
2550                 sleep(1);
2551         }
2552 }
2553
2554 static int ctdb_ctrl_modflags(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
2555                               struct ctdb_client_context *client,
2556                               uint32_t destnode, struct timeval timeout,
2557                               uint32_t set, uint32_t clear)
2558 {
2559         struct ctdb_node_map *nodemap;
2560         struct ctdb_node_flag_change flag_change;
2561         struct ctdb_req_control request;
2562         uint32_t *pnn_list;
2563         int ret, count;
2564
2565         ret = ctdb_ctrl_get_nodemap(mem_ctx, ev, client, destnode,
2566                                     tevent_timeval_zero(), &nodemap);
2567         if (ret != 0) {
2568                 return ret;
2569         }
2570
2571         flag_change.pnn = destnode;
2572         flag_change.old_flags = nodemap->node[destnode].flags;
2573         flag_change.new_flags = flag_change.old_flags | set;
2574         flag_change.new_flags &= ~clear;
2575
2576         count = list_of_connected_nodes(nodemap, -1, mem_ctx, &pnn_list);
2577         if (count == -1) {
2578                 return ENOMEM;
2579         }
2580
2581         ctdb_req_control_modify_flags(&request, &flag_change);
2582         ret = ctdb_client_control_multi(mem_ctx, ev, client, pnn_list, count,
2583                                         tevent_timeval_zero(), &request,
2584                                         NULL, NULL);
2585         return ret;
2586 }
2587
2588 struct ipreallocate_state {
2589         int status;
2590         bool done;
2591 };
2592
2593 static void ipreallocate_handler(uint64_t srvid, TDB_DATA data,
2594                                  void *private_data)
2595 {
2596         struct ipreallocate_state *state =
2597                 (struct ipreallocate_state *)private_data;
2598
2599         if (data.dsize != sizeof(int)) {
2600                 /* Ignore packet */
2601                 return;
2602         }
2603
2604         state->status = *(int *)data.dptr;
2605         state->done = true;
2606 }
2607
2608 static int ipreallocate(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb)
2609 {
2610         struct ctdb_srvid_message msg = { 0 };
2611         struct ipreallocate_state state;
2612         int ret;
2613
2614         msg.pnn = ctdb->pnn;
2615         msg.srvid = next_srvid(ctdb);
2616
2617         state.done = false;
2618         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
2619                                               msg.srvid,
2620                                               ipreallocate_handler, &state);
2621         if (ret != 0) {
2622                 return ret;
2623         }
2624
2625         while (true) {
2626                 ret = ctdb_message_takeover_run(mem_ctx, ctdb->ev,
2627                                                 ctdb->client,
2628                                                 CTDB_BROADCAST_CONNECTED,
2629                                                 &msg);
2630                 if (ret != 0) {
2631                         goto fail;
2632                 }
2633
2634                 ret = ctdb_client_wait_timeout(ctdb->ev, &state.done,
2635                                                TIMEOUT());
2636                 if (ret != 0) {
2637                         continue;
2638                 }
2639
2640                 if (state.status >= 0) {
2641                         ret = 0;
2642                 } else {
2643                         ret = state.status;
2644                 }
2645                 break;
2646         }
2647
2648 fail:
2649         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
2650                                            msg.srvid, &state);
2651         return ret;
2652 }
2653
2654 static int control_disable(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2655                            int argc, const char **argv)
2656 {
2657         int ret;
2658
2659         if (argc != 0) {
2660                 usage("disable");
2661         }
2662
2663         ret = check_flags(mem_ctx, ctdb, "disabled",
2664                           NODE_FLAGS_PERMANENTLY_DISABLED, true);
2665         if (ret == 0) {
2666                 return 0;
2667         }
2668
2669         ret = ctdb_ctrl_modflags(mem_ctx, ctdb->ev, ctdb->client,
2670                                  ctdb->cmd_pnn, TIMEOUT(),
2671                                  NODE_FLAGS_PERMANENTLY_DISABLED, 0);
2672         if (ret != 0) {
2673                 fprintf(stderr,
2674                         "Failed to set DISABLED flag on node %u\n",
2675                         ctdb->cmd_pnn);
2676                 return ret;
2677         }
2678
2679         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_PERMANENTLY_DISABLED, true);
2680         return ipreallocate(mem_ctx, ctdb);
2681 }
2682
2683 static int control_enable(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2684                           int argc, const char **argv)
2685 {
2686         int ret;
2687
2688         if (argc != 0) {
2689                 usage("enable");
2690         }
2691
2692         ret = check_flags(mem_ctx, ctdb, "disabled",
2693                           NODE_FLAGS_PERMANENTLY_DISABLED, false);
2694         if (ret == 0) {
2695                 return 0;
2696         }
2697
2698         ret = ctdb_ctrl_modflags(mem_ctx, ctdb->ev, ctdb->client,
2699                                  ctdb->cmd_pnn, TIMEOUT(),
2700                                  0, NODE_FLAGS_PERMANENTLY_DISABLED);
2701         if (ret != 0) {
2702                 fprintf(stderr, "Failed to reset DISABLED flag on node %u\n",
2703                         ctdb->cmd_pnn);
2704                 return ret;
2705         }
2706
2707         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_PERMANENTLY_DISABLED, false);
2708         return ipreallocate(mem_ctx, ctdb);
2709 }
2710
2711 static int control_stop(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2712                         int argc, const char **argv)
2713 {
2714         int ret;
2715
2716         if (argc != 0) {
2717                 usage("stop");
2718         }
2719
2720         ret = check_flags(mem_ctx, ctdb, "stopped",
2721                           NODE_FLAGS_STOPPED, true);
2722         if (ret == 0) {
2723                 return 0;
2724         }
2725
2726         ret = ctdb_ctrl_stop_node(mem_ctx, ctdb->ev, ctdb->client,
2727                                   ctdb->cmd_pnn, TIMEOUT());
2728         if (ret != 0) {
2729                 fprintf(stderr, "Failed to stop node %u\n", ctdb->cmd_pnn);
2730                 return ret;
2731         }
2732
2733         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_STOPPED, true);
2734         return ipreallocate(mem_ctx, ctdb);
2735 }
2736
2737 static int control_continue(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2738                             int argc, const char **argv)
2739 {
2740         int ret;
2741
2742         if (argc != 0) {
2743                 usage("continue");
2744         }
2745
2746         ret = check_flags(mem_ctx, ctdb, "stopped",
2747                           NODE_FLAGS_STOPPED, false);
2748         if (ret == 0) {
2749                 return 0;
2750         }
2751
2752         ret = ctdb_ctrl_continue_node(mem_ctx, ctdb->ev, ctdb->client,
2753                                       ctdb->cmd_pnn, TIMEOUT());
2754         if (ret != 0) {
2755                 fprintf(stderr, "Failed to continue stopped node %u\n",
2756                         ctdb->cmd_pnn);
2757                 return ret;
2758         }
2759
2760         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_STOPPED, false);
2761         return ipreallocate(mem_ctx, ctdb);
2762 }
2763
2764 static int control_ban(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2765                        int argc, const char **argv)
2766 {
2767         struct ctdb_ban_state ban_state;
2768         int ret;
2769
2770         if (argc != 1) {
2771                 usage("ban");
2772         }
2773
2774         ret = check_flags(mem_ctx, ctdb, "banned",
2775                           NODE_FLAGS_BANNED, true);
2776         if (ret == 0) {
2777                 return 0;
2778         }
2779
2780         ban_state.pnn = ctdb->cmd_pnn;
2781         ban_state.time = strtoul(argv[0], NULL, 0);
2782
2783         if (ban_state.time == 0) {
2784                 fprintf(stderr, "Ban time cannot be zero\n");
2785                 return EINVAL;
2786         }
2787
2788         ret = ctdb_ctrl_set_ban_state(mem_ctx, ctdb->ev, ctdb->client,
2789                                       ctdb->cmd_pnn, TIMEOUT(), &ban_state);
2790         if (ret != 0) {
2791                 fprintf(stderr, "Failed to ban node %u\n", ctdb->cmd_pnn);
2792                 return ret;
2793         }
2794
2795         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_BANNED, true);
2796         return ipreallocate(mem_ctx, ctdb);
2797
2798 }
2799
2800 static int control_unban(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2801                          int argc, const char **argv)
2802 {
2803         struct ctdb_ban_state ban_state;
2804         int ret;
2805
2806         if (argc != 0) {
2807                 usage("unban");
2808         }
2809
2810         ret = check_flags(mem_ctx, ctdb, "banned",
2811                           NODE_FLAGS_BANNED, false);
2812         if (ret == 0) {
2813                 return 0;
2814         }
2815
2816         ban_state.pnn = ctdb->cmd_pnn;
2817         ban_state.time = 0;
2818
2819         ret = ctdb_ctrl_set_ban_state(mem_ctx, ctdb->ev, ctdb->client,
2820                                       ctdb->cmd_pnn, TIMEOUT(), &ban_state);
2821         if (ret != 0) {
2822                 fprintf(stderr, "Failed to unban node %u\n", ctdb->cmd_pnn);
2823                 return ret;
2824         }
2825
2826         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_BANNED, false);
2827         return ipreallocate(mem_ctx, ctdb);
2828
2829 }
2830
2831 static int control_shutdown(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2832                             int argc, const char **argv)
2833 {
2834         int ret;
2835
2836         if (argc != 0) {
2837                 usage("shutdown");
2838         }
2839
2840         ret = ctdb_ctrl_shutdown(mem_ctx, ctdb->ev, ctdb->client,
2841                                  ctdb->cmd_pnn, TIMEOUT());
2842         if (ret != 0) {
2843                 fprintf(stderr, "Unable to shutdown node %u\n", ctdb->cmd_pnn);
2844                 return ret;
2845         }
2846
2847         return 0;
2848 }
2849
2850 static int get_generation(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2851                           uint32_t *generation)
2852 {
2853         uint32_t recmaster;
2854         int recmode;
2855         struct ctdb_vnn_map *vnnmap;
2856         int ret;
2857
2858 again:
2859         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
2860                                       ctdb->cmd_pnn, TIMEOUT(), &recmaster);
2861         if (ret != 0) {
2862                 fprintf(stderr, "Failed to find recovery master\n");
2863                 return ret;
2864         }
2865
2866         ret = ctdb_ctrl_get_recmode(mem_ctx, ctdb->ev, ctdb->client,
2867                                     recmaster, TIMEOUT(), &recmode);
2868         if (ret != 0) {
2869                 fprintf(stderr, "Failed to get recovery mode from node %u\n",
2870                         recmaster);
2871                 return ret;
2872         }
2873
2874         if (recmode == CTDB_RECOVERY_ACTIVE) {
2875                 sleep(1);
2876                 goto again;
2877         }
2878
2879         ret = ctdb_ctrl_getvnnmap(mem_ctx, ctdb->ev, ctdb->client,
2880                                   recmaster, TIMEOUT(), &vnnmap);
2881         if (ret != 0) {
2882                 fprintf(stderr, "Failed to get generation from node %u\n",
2883                         recmaster);
2884                 return ret;
2885         }
2886
2887         if (vnnmap->generation == INVALID_GENERATION) {
2888                 talloc_free(vnnmap);
2889                 sleep(1);
2890                 goto again;
2891         }
2892
2893         *generation = vnnmap->generation;
2894         talloc_free(vnnmap);
2895         return 0;
2896 }
2897
2898
2899 static int control_recover(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2900                            int argc, const char **argv)
2901 {
2902         uint32_t generation, next_generation;
2903         int ret;
2904
2905         if (argc != 0) {
2906                 usage("recover");
2907         }
2908
2909         ret = get_generation(mem_ctx, ctdb, &generation);
2910         if (ret != 0) {
2911                 return ret;
2912         }
2913
2914         ret = ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
2915                                     ctdb->cmd_pnn, TIMEOUT(),
2916                                     CTDB_RECOVERY_ACTIVE);
2917         if (ret != 0) {
2918                 fprintf(stderr, "Failed to set recovery mode active\n");
2919                 return ret;
2920         }
2921
2922         while (1) {
2923                 ret = get_generation(mem_ctx, ctdb, &next_generation);
2924                 if (ret != 0) {
2925                         fprintf(stderr,
2926                                 "Failed to confirm end of recovery\n");
2927                         return ret;
2928                 }
2929
2930                 if (next_generation != generation) {
2931                         break;
2932                 }
2933
2934                 sleep (1);
2935         }
2936
2937         return 0;
2938 }
2939
2940 static int control_ipreallocate(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2941                                 int argc, const char **argv)
2942 {
2943         if (argc != 0) {
2944                 usage("ipreallocate");
2945         }
2946
2947         return ipreallocate(mem_ctx, ctdb);
2948 }
2949
2950 static int control_isnotrecmaster(TALLOC_CTX *mem_ctx,
2951                                   struct ctdb_context *ctdb,
2952                                   int argc, const char **argv)
2953 {
2954         uint32_t recmaster;
2955         int ret;
2956
2957         if (argc != 0) {
2958                 usage("isnotrecmaster");
2959         }
2960
2961         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
2962                                       ctdb->pnn, TIMEOUT(), &recmaster);
2963         if (ret != 0) {
2964                 fprintf(stderr, "Failed to get recmaster\n");
2965                 return ret;
2966         }
2967
2968         if (recmaster != ctdb->pnn) {
2969                 printf("this node is not the recmaster\n");
2970                 return 1;
2971         }
2972
2973         printf("this node is the recmaster\n");
2974         return 0;
2975 }
2976
2977 static int control_gratarp(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2978                            int argc, const char **argv)
2979 {
2980         struct ctdb_addr_info addr_info;
2981         int ret;
2982
2983         if (argc != 2) {
2984                 usage("gratarp");
2985         }
2986
2987         if (! parse_ip(argv[0], NULL, 0, &addr_info.addr)) {
2988                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
2989                 return 1;
2990         }
2991         addr_info.iface = argv[1];
2992
2993         ret = ctdb_ctrl_send_gratuitous_arp(mem_ctx, ctdb->ev, ctdb->client,
2994                                             ctdb->cmd_pnn, TIMEOUT(),
2995                                             &addr_info);
2996         if (ret != 0) {
2997                 fprintf(stderr, "Unable to send gratuitous arp from node %u\n",
2998                         ctdb->cmd_pnn);
2999                 return ret;
3000         }
3001
3002         return 0;
3003 }
3004
3005 static int control_tickle(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3006                            int argc, const char **argv)
3007 {
3008         ctdb_sock_addr src, dst;
3009         int ret;
3010
3011         if (argc != 0 && argc != 2) {
3012                 usage("tickle");
3013         }
3014
3015         if (argc == 0) {
3016                 struct ctdb_connection *clist;
3017                 int count;
3018                 int i, num_failed;
3019
3020                 ret = ctdb_parse_connections(stdin, mem_ctx, &count, &clist);
3021                 if (ret != 0) {
3022                         return ret;
3023                 }
3024
3025                 num_failed = 0;
3026                 for (i=0; i<count; i++) {
3027                         ret = ctdb_sys_send_tcp(&clist[i].src,
3028                                                 &clist[i].dst,
3029                                                 0, 0, 0);
3030                         if (ret != 0) {
3031                                 num_failed += 1;
3032                         }
3033                 }
3034
3035                 TALLOC_FREE(clist);
3036
3037                 if (num_failed > 0) {
3038                         fprintf(stderr, "Failed to send %d tickles\n",
3039                                 num_failed);
3040                         return 1;
3041                 }
3042
3043                 return 0;
3044         }
3045
3046
3047         if (! parse_ip_port(argv[0], &src)) {
3048                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3049                 return 1;
3050         }
3051
3052         if (! parse_ip_port(argv[1], &dst)) {
3053                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3054                 return 1;
3055         }
3056
3057         ret = ctdb_sys_send_tcp(&src, &dst, 0, 0, 0);
3058         if (ret != 0) {
3059                 fprintf(stderr, "Failed to send tickle ack\n");
3060                 return ret;
3061         }
3062
3063         return 0;
3064 }
3065
3066 static int control_gettickles(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3067                               int argc, const char **argv)
3068 {
3069         ctdb_sock_addr addr;
3070         struct ctdb_tickle_list *tickles;
3071         unsigned port = 0;
3072         int ret, i;
3073
3074         if (argc < 1 || argc > 2) {
3075                 usage("gettickles");
3076         }
3077
3078         if (argc == 2) {
3079                 port = strtoul(argv[1], NULL, 10);
3080         }
3081
3082         if (! parse_ip(argv[0], NULL, port, &addr)) {
3083                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3084                 return 1;
3085         }
3086
3087         ret = ctdb_ctrl_get_tcp_tickle_list(mem_ctx, ctdb->ev, ctdb->client,
3088                                             ctdb->cmd_pnn, TIMEOUT(), &addr,
3089                                             &tickles);
3090         if (ret != 0) {
3091                 fprintf(stderr, "Failed to get list of connections\n");
3092                 return ret;
3093         }
3094
3095         if (options.machinereadable) {
3096                 printf("%s%s%s%s%s%s%s%s%s\n",
3097                        options.sep,
3098                        "Source IP", options.sep,
3099                        "Port", options.sep,
3100                        "Destiation IP", options.sep,
3101                        "Port", options.sep);
3102                 for (i=0; i<tickles->num; i++) {
3103                         printf("%s%s%s%u%s%s%s%u%s\n", options.sep,
3104                                ctdb_sock_addr_to_string(
3105                                        mem_ctx, &tickles->conn[i].src, false),
3106                                options.sep,
3107                                ntohs(tickles->conn[i].src.ip.sin_port),
3108                                options.sep,
3109                                ctdb_sock_addr_to_string(
3110                                        mem_ctx, &tickles->conn[i].dst, false),
3111                                options.sep,
3112                                ntohs(tickles->conn[i].dst.ip.sin_port),
3113                                options.sep);
3114                 }
3115         } else {
3116                 printf("Connections for IP: %s\n",
3117                        ctdb_sock_addr_to_string(mem_ctx,
3118                                                 &tickles->addr, false));
3119                 printf("Num connections: %u\n", tickles->num);
3120                 for (i=0; i<tickles->num; i++) {
3121                         printf("SRC: %s   DST: %s\n",
3122                                ctdb_sock_addr_to_string(
3123                                        mem_ctx, &tickles->conn[i].src, true),
3124                                ctdb_sock_addr_to_string(
3125                                        mem_ctx, &tickles->conn[i].dst, true));
3126                 }
3127         }
3128
3129         talloc_free(tickles);
3130         return 0;
3131 }
3132
3133 typedef void (*clist_request_func)(struct ctdb_req_control *request,
3134                                    struct ctdb_connection *conn);
3135
3136 typedef int (*clist_reply_func)(struct ctdb_reply_control *reply);
3137
3138 struct process_clist_state {
3139         struct ctdb_connection *clist;
3140         int count;
3141         int num_failed, num_total;
3142         clist_reply_func reply_func;
3143 };
3144
3145 static void process_clist_done(struct tevent_req *subreq);
3146
3147 static struct tevent_req *process_clist_send(
3148                                         TALLOC_CTX *mem_ctx,
3149                                         struct ctdb_context *ctdb,
3150                                         struct ctdb_connection *clist,
3151                                         int count,
3152                                         clist_request_func request_func,
3153                                         clist_reply_func reply_func)
3154 {
3155         struct tevent_req *req, *subreq;
3156         struct process_clist_state *state;
3157         struct ctdb_req_control request;
3158         int i;
3159
3160         req = tevent_req_create(mem_ctx, &state, struct process_clist_state);
3161         if (req == NULL) {
3162                 return NULL;
3163         }
3164
3165         state->clist = clist;
3166         state->count = count;
3167         state->reply_func = reply_func;
3168
3169         for (i=0; i<count; i++) {
3170                 request_func(&request, &clist[i]);
3171                 subreq = ctdb_client_control_send(state, ctdb->ev,
3172                                                   ctdb->client, ctdb->cmd_pnn,
3173                                                   TIMEOUT(), &request);
3174                 if (tevent_req_nomem(subreq, req)) {
3175                         return tevent_req_post(req, ctdb->ev);
3176                 }
3177                 tevent_req_set_callback(subreq, process_clist_done, req);
3178         }
3179
3180         return req;
3181 }
3182
3183 static void process_clist_done(struct tevent_req *subreq)
3184 {
3185         struct tevent_req *req = tevent_req_callback_data(
3186                 subreq, struct tevent_req);
3187         struct process_clist_state *state = tevent_req_data(
3188                 req, struct process_clist_state);
3189         struct ctdb_reply_control *reply;
3190         int ret;
3191         bool status;
3192
3193         status = ctdb_client_control_recv(subreq, NULL, state, &reply);
3194         TALLOC_FREE(subreq);
3195         if (! status) {
3196                 state->num_failed += 1;
3197                 goto done;
3198         }
3199
3200         ret = state->reply_func(reply);
3201         if (ret != 0) {
3202                 state->num_failed += 1;
3203                 goto done;
3204         }
3205
3206 done:
3207         state->num_total += 1;
3208         if (state->num_total == state->count) {
3209                 tevent_req_done(req);
3210         }
3211 }
3212
3213 static int process_clist_recv(struct tevent_req *req)
3214 {
3215         struct process_clist_state *state = tevent_req_data(
3216                 req, struct process_clist_state);
3217
3218         return state->num_failed;
3219 }
3220
3221 static int control_addtickle(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3222                              int argc, const char **argv)
3223 {
3224         struct ctdb_connection conn;
3225         int ret;
3226
3227         if (argc != 0 && argc != 2) {
3228                 usage("addtickle");
3229         }
3230
3231         if (argc == 0) {
3232                 struct ctdb_connection *clist;
3233                 struct tevent_req *req;
3234                 int count;
3235
3236                 ret = ctdb_parse_connections(stdin, mem_ctx, &count, &clist);
3237                 if (ret != 0) {
3238                         return ret;
3239                 }
3240                 if (count == 0) {
3241                         return 0;
3242                 }
3243
3244                 req = process_clist_send(mem_ctx, ctdb, clist, count,
3245                                  ctdb_req_control_tcp_add_delayed_update,
3246                                  ctdb_reply_control_tcp_add_delayed_update);
3247                 if (req == NULL) {
3248                         talloc_free(clist);
3249                         return ENOMEM;
3250                 }
3251
3252                 tevent_req_poll(req, ctdb->ev);
3253                 talloc_free(clist);
3254
3255                 ret = process_clist_recv(req);
3256                 if (ret != 0) {
3257                         fprintf(stderr, "Failed to add %d tickles\n", ret);
3258                         return 1;
3259                 }
3260
3261                 return 0;
3262         }
3263
3264         if (! parse_ip_port(argv[0], &conn.src)) {
3265                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3266                 return 1;
3267         }
3268         if (! parse_ip_port(argv[1], &conn.dst)) {
3269                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3270                 return 1;
3271         }
3272
3273         ret = ctdb_ctrl_tcp_add_delayed_update(mem_ctx, ctdb->ev,
3274                                                ctdb->client, ctdb->cmd_pnn,
3275                                                TIMEOUT(), &conn);
3276         if (ret != 0) {
3277                 fprintf(stderr, "Failed to register connection\n");
3278                 return ret;
3279         }
3280
3281         return 0;
3282 }
3283
3284 static int control_deltickle(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3285                              int argc, const char **argv)
3286 {
3287         struct ctdb_connection conn;
3288         int ret;
3289
3290         if (argc != 0 && argc != 2) {
3291                 usage("deltickle");
3292         }
3293
3294         if (argc == 0) {
3295                 struct ctdb_connection *clist;
3296                 struct tevent_req *req;
3297                 int count;
3298
3299                 ret = ctdb_parse_connections(stdin, mem_ctx, &count, &clist);
3300                 if (ret != 0) {
3301                         return ret;
3302                 }
3303                 if (count == 0) {
3304                         return 0;
3305                 }
3306
3307                 req = process_clist_send(mem_ctx, ctdb, clist, count,
3308                                          ctdb_req_control_tcp_remove,
3309                                          ctdb_reply_control_tcp_remove);
3310                 if (req == NULL) {
3311                         talloc_free(clist);
3312                         return ENOMEM;
3313                 }
3314
3315                 tevent_req_poll(req, ctdb->ev);
3316                 talloc_free(clist);
3317
3318                 ret = process_clist_recv(req);
3319                 if (ret != 0) {
3320                         fprintf(stderr, "Failed to remove %d tickles\n", ret);
3321                         return 1;
3322                 }
3323
3324                 return 0;
3325         }
3326
3327         if (! parse_ip_port(argv[0], &conn.src)) {
3328                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3329                 return 1;
3330         }
3331         if (! parse_ip_port(argv[1], &conn.dst)) {
3332                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3333                 return 1;
3334         }
3335
3336         ret = ctdb_ctrl_tcp_remove(mem_ctx, ctdb->ev, ctdb->client,
3337                                    ctdb->cmd_pnn, TIMEOUT(), &conn);
3338         if (ret != 0) {
3339                 fprintf(stderr, "Failed to unregister connection\n");
3340                 return ret;
3341         }
3342
3343         return 0;
3344 }
3345
3346 static int control_listnodes(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3347                              int argc, const char **argv)
3348 {
3349         struct ctdb_node_map *nodemap;
3350         int i;
3351
3352         if (argc != 0) {
3353                 usage("listnodes");
3354         }
3355
3356         nodemap = read_nodes_file(mem_ctx, CTDB_UNKNOWN_PNN);
3357         if (nodemap == NULL) {
3358                 return 1;
3359         }
3360
3361         for (i=0; i<nodemap->num; i++) {
3362                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
3363                         continue;
3364                 }
3365
3366                 if (options.machinereadable) {
3367                         printf("%s%u%s%s%s\n", options.sep,
3368                                nodemap->node[i].pnn, options.sep,
3369                                ctdb_sock_addr_to_string(
3370                                        mem_ctx, &nodemap->node[i].addr, false),
3371                                options.sep);
3372                 } else {
3373                         printf("%s\n",
3374                                ctdb_sock_addr_to_string(
3375                                        mem_ctx, &nodemap->node[i].addr, false));
3376                 }
3377         }
3378
3379         return 0;
3380 }
3381
3382 static bool nodemap_identical(struct ctdb_node_map *nodemap1,
3383                               struct ctdb_node_map *nodemap2)
3384 {
3385         int i;
3386
3387         if (nodemap1->num != nodemap2->num) {
3388                 return false;
3389         }
3390
3391         for (i=0; i<nodemap1->num; i++) {
3392                 struct ctdb_node_and_flags *n1, *n2;
3393
3394                 n1 = &nodemap1->node[i];
3395                 n2 = &nodemap2->node[i];
3396
3397                 if ((n1->pnn != n2->pnn) ||
3398                     (n1->flags != n2->flags) ||
3399                     ! ctdb_sock_addr_same_ip(&n1->addr, &n2->addr)) {
3400                         return false;
3401                 }
3402         }
3403
3404         return true;
3405 }
3406
3407 static int check_node_file_changes(TALLOC_CTX *mem_ctx,
3408                                    struct ctdb_node_map *nm,
3409                                    struct ctdb_node_map *fnm,
3410                                    bool *reload)
3411 {
3412         int i;
3413         bool check_failed = false;
3414
3415         *reload = false;
3416
3417         for (i=0; i<nm->num; i++) {
3418                 if (i >= fnm->num) {
3419                         fprintf(stderr,
3420                                 "Node %u (%s) missing from nodes file\n",
3421                                 nm->node[i].pnn,
3422                                 ctdb_sock_addr_to_string(
3423                                         mem_ctx, &nm->node[i].addr, false));
3424                         check_failed = true;
3425                         continue;
3426                 }
3427                 if (nm->node[i].flags & NODE_FLAGS_DELETED &&
3428                     fnm->node[i].flags & NODE_FLAGS_DELETED) {
3429                         /* Node remains deleted */
3430                         continue;
3431                 }
3432
3433                 if (! (nm->node[i].flags & NODE_FLAGS_DELETED) &&
3434                     ! (fnm->node[i].flags & NODE_FLAGS_DELETED)) {
3435                         /* Node not newly nor previously deleted */
3436                         if (! ctdb_same_ip(&nm->node[i].addr,
3437                                            &fnm->node[i].addr)) {
3438                                 fprintf(stderr,
3439                                         "Node %u has changed IP address"
3440                                         " (was %s, now %s)\n",
3441                                         nm->node[i].pnn,
3442                                         ctdb_sock_addr_to_string(
3443                                                 mem_ctx,
3444                                                 &nm->node[i].addr, false),
3445                                         ctdb_sock_addr_to_string(
3446                                                 mem_ctx,
3447                                                 &fnm->node[i].addr, false));
3448                                 check_failed = true;
3449                         } else {
3450                                 if (nm->node[i].flags & NODE_FLAGS_DISCONNECTED) {
3451                                         fprintf(stderr,
3452                                                 "WARNING: Node %u is disconnected."
3453                                                 " You MUST fix this node manually!\n",
3454                                                 nm->node[i].pnn);
3455                                 }
3456                         }
3457                         continue;
3458                 }
3459
3460                 if (fnm->node[i].flags & NODE_FLAGS_DELETED) {
3461                         /* Node is being deleted */
3462                         printf("Node %u is DELETED\n", nm->node[i].pnn);
3463                         *reload = true;
3464                         if (! (nm->node[i].flags & NODE_FLAGS_DISCONNECTED)) {
3465                                 fprintf(stderr,
3466                                         "ERROR: Node %u is still connected\n",
3467                                         nm->node[i].pnn);
3468                                 check_failed = true;
3469                         }
3470                         continue;
3471                 }
3472
3473                 if (nm->node[i].flags & NODE_FLAGS_DELETED) {
3474                         /* Node was previously deleted */
3475                         printf("Node %u is UNDELETED\n", nm->node[i].pnn);
3476                         *reload = true;
3477                 }
3478         }
3479
3480         if (check_failed) {
3481                 fprintf(stderr,
3482                         "ERROR: Nodes will not be reloaded due to previous error\n");
3483                 return 1;
3484         }
3485
3486         /* Leftover nodes in file are NEW */
3487         for (; i < fnm->num; i++) {
3488                 printf("Node %u is NEW\n", fnm->node[i].pnn);
3489                 *reload = true;
3490         }
3491
3492         return 0;
3493 }
3494
3495 struct disable_recoveries_state {
3496         uint32_t *pnn_list;
3497         int node_count;
3498         bool *reply;
3499         int status;
3500         bool done;
3501 };
3502
3503 static void disable_recoveries_handler(uint64_t srvid, TDB_DATA data,
3504                                        void *private_data)
3505 {
3506         struct disable_recoveries_state *state =
3507                 (struct disable_recoveries_state *)private_data;
3508         int ret, i;
3509
3510         if (data.dsize != sizeof(int)) {
3511                 /* Ignore packet */
3512                 return;
3513         }
3514
3515         /* ret will be a PNN (i.e. >=0) on success, or negative on error */
3516         ret = *(int *)data.dptr;
3517         if (ret < 0) {
3518                 state->status = ret;
3519                 state->done = true;
3520                 return;
3521         }
3522         for (i=0; i<state->node_count; i++) {
3523                 if (state->pnn_list[i] == ret) {
3524                         state->reply[i] = true;
3525                         break;
3526                 }
3527         }
3528
3529         state->done = true;
3530         for (i=0; i<state->node_count; i++) {
3531                 if (! state->reply[i]) {
3532                         state->done = false;
3533                         break;
3534                 }
3535         }
3536 }
3537
3538 static int disable_recoveries(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3539                               uint32_t timeout, uint32_t *pnn_list, int count)
3540 {
3541         struct ctdb_disable_message disable = { 0 };
3542         struct disable_recoveries_state state;
3543         int ret, i;
3544
3545         disable.pnn = ctdb->pnn;
3546         disable.srvid = next_srvid(ctdb);
3547         disable.timeout = timeout;
3548
3549         state.pnn_list = pnn_list;
3550         state.node_count = count;
3551         state.done = false;
3552         state.status = 0;
3553         state.reply = talloc_zero_array(mem_ctx, bool, count);
3554         if (state.reply == NULL) {
3555                 return ENOMEM;
3556         }
3557
3558         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
3559                                               disable.srvid,
3560                                               disable_recoveries_handler,
3561                                               &state);
3562         if (ret != 0) {
3563                 return ret;
3564         }
3565
3566         for (i=0; i<count; i++) {
3567                 ret = ctdb_message_disable_recoveries(mem_ctx, ctdb->ev,
3568                                                       ctdb->client,
3569                                                       pnn_list[i],
3570                                                       &disable);
3571                 if (ret != 0) {
3572                         goto fail;
3573                 }
3574         }
3575
3576         ret = ctdb_client_wait_timeout(ctdb->ev, &state.done, TIMEOUT());
3577         if (ret == ETIME) {
3578                 fprintf(stderr, "Timed out waiting to disable recoveries\n");
3579         } else {
3580                 ret = (state.status >= 0 ? 0 : 1);
3581         }
3582
3583 fail:
3584         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
3585                                            disable.srvid, &state);
3586         return ret;
3587 }
3588
3589 static int control_reloadnodes(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3590                                int argc, const char **argv)
3591 {
3592         struct ctdb_node_map *nodemap = NULL;
3593         struct ctdb_node_map *file_nodemap;
3594         struct ctdb_node_map *remote_nodemap;
3595         struct ctdb_req_control request;
3596         struct ctdb_reply_control **reply;
3597         bool reload;
3598         int ret, i;
3599         uint32_t *pnn_list;
3600         int count;
3601
3602         nodemap = get_nodemap(ctdb, false);
3603         if (nodemap == NULL) {
3604                 return 1;
3605         }
3606
3607         file_nodemap = read_nodes_file(mem_ctx, ctdb->pnn);
3608         if (file_nodemap == NULL) {
3609                 return 1;
3610         }
3611
3612         for (i=0; i<nodemap->num; i++) {
3613                 if (nodemap->node[i].flags & NODE_FLAGS_DISCONNECTED) {
3614                         continue;
3615                 }
3616
3617                 ret = ctdb_ctrl_get_nodes_file(mem_ctx, ctdb->ev, ctdb->client,
3618                                                nodemap->node[i].pnn, TIMEOUT(),
3619                                                &remote_nodemap);
3620                 if (ret != 0) {
3621                         fprintf(stderr,
3622                                 "ERROR: Failed to get nodes file from node %u\n",
3623                                 nodemap->node[i].pnn);
3624                         return ret;
3625                 }
3626
3627                 if (! nodemap_identical(file_nodemap, remote_nodemap)) {
3628                         fprintf(stderr,
3629                                 "ERROR: Nodes file on node %u differs"
3630                                  " from current node (%u)\n",
3631                                  nodemap->node[i].pnn, ctdb->pnn);
3632                         return 1;
3633                 }
3634         }
3635
3636         ret = check_node_file_changes(mem_ctx, nodemap, file_nodemap, &reload);
3637         if (ret != 0) {
3638                 return ret;
3639         }
3640
3641         if (! reload) {
3642                 fprintf(stderr, "No change in nodes file,"
3643                                 " skipping unnecessary reload\n");
3644                 return 0;
3645         }
3646
3647         count = list_of_connected_nodes(nodemap, CTDB_UNKNOWN_PNN,
3648                                         mem_ctx, &pnn_list);
3649         if (count <= 0) {
3650                 fprintf(stderr, "Memory allocation error\n");
3651                 return 1;
3652         }
3653
3654         ret = disable_recoveries(mem_ctx, ctdb, 2*options.timelimit,
3655                                  pnn_list, count);
3656         if (ret != 0) {
3657                 fprintf(stderr, "Failed to disable recoveries\n");
3658                 return ret;
3659         }
3660
3661         ctdb_req_control_reload_nodes_file(&request);
3662         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
3663                                         pnn_list, count, TIMEOUT(),
3664                                         &request, NULL, &reply);
3665         if (ret != 0) {
3666                 bool failed = false;
3667
3668                 for (i=0; i<count; i++) {
3669                         ret = ctdb_reply_control_reload_nodes_file(reply[i]);
3670                         if (ret != 0) {
3671                                 fprintf(stderr,
3672                                         "Node %u failed to reload nodes\n",
3673                                         pnn_list[i]);
3674                                 failed = true;
3675                         }
3676                 }
3677                 if (failed) {
3678                         fprintf(stderr,
3679                                 "You MUST fix failed nodes manually!\n");
3680                 }
3681         }
3682
3683         ret = disable_recoveries(mem_ctx, ctdb, 0, pnn_list, count);
3684         if (ret != 0) {
3685                 fprintf(stderr, "Failed to enable recoveries\n");
3686                 return ret;
3687         }
3688
3689         return 0;
3690 }
3691
3692 static int moveip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3693                   ctdb_sock_addr *addr, uint32_t pnn)
3694 {
3695         struct ctdb_public_ip_list *pubip_list;
3696         struct ctdb_public_ip pubip;
3697         struct ctdb_node_map *nodemap;
3698         struct ctdb_req_control request;
3699         uint32_t *pnn_list;
3700         int ret, i, count;
3701
3702         ret = ctdb_message_disable_ip_check(mem_ctx, ctdb->ev, ctdb->client,
3703                                             CTDB_BROADCAST_CONNECTED,
3704                                             2*options.timelimit);
3705         if (ret != 0) {
3706                 fprintf(stderr, "Failed to disable IP check\n");
3707                 return ret;
3708         }
3709
3710         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3711                                        pnn, TIMEOUT(), false, &pubip_list);
3712         if (ret != 0) {
3713                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3714                         pnn);
3715                 return ret;
3716         }
3717
3718         for (i=0; i<pubip_list->num; i++) {
3719                 if (ctdb_same_ip(addr, &pubip_list->ip[i].addr)) {
3720                         break;
3721                 }
3722         }
3723
3724         if (i == pubip_list->num) {
3725                 fprintf(stderr, "Node %u CANNOT host IP address %s\n",
3726                         pnn, ctdb_sock_addr_to_string(mem_ctx, addr, false));
3727                 return 1;
3728         }
3729
3730         nodemap = get_nodemap(ctdb, false);
3731         if (nodemap == NULL) {
3732                 return 1;
3733         }
3734
3735         count = list_of_active_nodes(nodemap, pnn, mem_ctx, &pnn_list);
3736         if (count <= 0) {
3737                 fprintf(stderr, "Memory allocation error\n");
3738                 return 1;
3739         }
3740
3741         pubip.pnn = pnn;
3742         pubip.addr = *addr;
3743         ctdb_req_control_release_ip(&request, &pubip);
3744
3745         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
3746                                         pnn_list, count, TIMEOUT(),
3747                                         &request, NULL, NULL);
3748         if (ret != 0) {
3749                 fprintf(stderr, "Failed to release IP on nodes\n");
3750                 return ret;
3751         }
3752
3753         ret = ctdb_ctrl_takeover_ip(mem_ctx, ctdb->ev, ctdb->client,
3754                                     pnn, TIMEOUT(), &pubip);
3755         if (ret != 0) {
3756                 fprintf(stderr, "Failed to takeover IP on node %u\n", pnn);
3757                 return ret;
3758         }
3759
3760         return 0;
3761 }
3762
3763 static int control_moveip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3764                           int argc, const char **argv)
3765 {
3766         ctdb_sock_addr addr;
3767         uint32_t pnn;
3768         int ret, retries = 0;
3769
3770         if (argc != 2) {
3771                 usage("moveip");
3772         }
3773
3774         if (! parse_ip(argv[0], NULL, 0, &addr)) {
3775                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3776                 return 1;
3777         }
3778
3779         pnn = strtoul(argv[1], NULL, 10);
3780         if (pnn == CTDB_UNKNOWN_PNN) {
3781                 fprintf(stderr, "Invalid PNN %s\n", argv[1]);
3782                 return 1;
3783         }
3784
3785         while (retries < 5) {
3786                 ret = moveip(mem_ctx, ctdb, &addr, pnn);
3787                 if (ret == 0) {
3788                         break;
3789                 }
3790
3791                 sleep(3);
3792                 retries++;
3793         }
3794
3795         if (ret != 0) {
3796                 fprintf(stderr, "Failed to move IP %s to node %u\n",
3797                         argv[0], pnn);
3798                 return ret;
3799         }
3800
3801         return 0;
3802 }
3803
3804 static int rebalancenode(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3805                          uint32_t pnn)
3806 {
3807         int ret;
3808
3809         ret = ctdb_message_rebalance_node(mem_ctx, ctdb->ev, ctdb->client,
3810                                           CTDB_BROADCAST_CONNECTED, pnn);
3811         if (ret != 0) {
3812                 fprintf(stderr,
3813                         "Failed to ask recovery master to distribute IPs\n");
3814                 return ret;
3815         }
3816
3817         return 0;
3818 }
3819
3820 static int control_addip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3821                          int argc, const char **argv)
3822 {
3823         ctdb_sock_addr addr;
3824         struct ctdb_public_ip_list *pubip_list;
3825         struct ctdb_addr_info addr_info;
3826         unsigned int mask;
3827         int ret, i, retries = 0;
3828
3829         if (argc != 2) {
3830                 usage("addip");
3831         }
3832
3833         if (! parse_ip_mask(argv[0], argv[1], &addr, &mask)) {
3834                 fprintf(stderr, "Invalid IP/Mask %s\n", argv[0]);
3835                 return 1;
3836         }
3837
3838         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3839                                        ctdb->cmd_pnn, TIMEOUT(),
3840                                        false, &pubip_list);
3841         if (ret != 0) {
3842                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3843                         ctdb->cmd_pnn);
3844                 return 1;
3845         }
3846
3847         for (i=0; i<pubip_list->num; i++) {
3848                 if (ctdb_same_ip(&addr, &pubip_list->ip[i].addr)) {
3849                         fprintf(stderr, "Node already knows about IP %s\n",
3850                                 ctdb_sock_addr_to_string(mem_ctx,
3851                                                          &addr, false));
3852                         return 0;
3853                 }
3854         }
3855
3856         addr_info.addr = addr;
3857         addr_info.mask = mask;
3858         addr_info.iface = argv[1];
3859
3860         while (retries < 5) {
3861                 ret = ctdb_ctrl_add_public_ip(mem_ctx, ctdb->ev, ctdb->client,
3862                                               ctdb->cmd_pnn, TIMEOUT(),
3863                                               &addr_info);
3864                 if (ret == 0) {
3865                         break;
3866                 }
3867
3868                 sleep(3);
3869                 retries++;
3870         }
3871
3872         if (ret != 0) {
3873                 fprintf(stderr, "Failed to add public IP to node %u."
3874                                 " Giving up\n", ctdb->cmd_pnn);
3875                 return ret;
3876         }
3877
3878         ret = rebalancenode(mem_ctx, ctdb, ctdb->cmd_pnn);
3879         if (ret != 0) {
3880                 return ret;
3881         }
3882
3883         return 0;
3884 }
3885
3886 static int control_delip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3887                          int argc, const char **argv)
3888 {
3889         ctdb_sock_addr addr;
3890         struct ctdb_public_ip_list *pubip_list;
3891         struct ctdb_addr_info addr_info;
3892         int ret, i;
3893
3894         if (argc != 1) {
3895                 usage("delip");
3896         }
3897
3898         if (! parse_ip(argv[0], NULL, 0, &addr)) {
3899                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3900                 return 1;
3901         }
3902
3903         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3904                                        ctdb->cmd_pnn, TIMEOUT(),
3905                                        false, &pubip_list);
3906         if (ret != 0) {
3907                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3908                         ctdb->cmd_pnn);
3909                 return 1;
3910         }
3911
3912         for (i=0; i<pubip_list->num; i++) {
3913                 if (ctdb_same_ip(&addr, &pubip_list->ip[i].addr)) {
3914                         break;
3915                 }
3916         }
3917
3918         if (i == pubip_list->num) {
3919                 fprintf(stderr, "Node does not know about IP address %s\n",
3920                         ctdb_sock_addr_to_string(mem_ctx, &addr, false));
3921                 return 0;
3922         }
3923
3924         addr_info.addr = addr;
3925         addr_info.mask = 0;
3926         addr_info.iface = NULL;
3927
3928         ret = ctdb_ctrl_del_public_ip(mem_ctx, ctdb->ev, ctdb->client,
3929                                       ctdb->cmd_pnn, TIMEOUT(), &addr_info);
3930         if (ret != 0) {
3931                 fprintf(stderr, "Failed to delete public IP from node %u\n",
3932                         ctdb->cmd_pnn);
3933                 return ret;
3934         }
3935
3936         return 0;
3937 }
3938
3939 #define DB_VERSION      3
3940 #define MAX_DB_NAME     64
3941 #define MAX_REC_BUFFER_SIZE     (100*1000)
3942
3943 struct db_header {
3944         unsigned long version;
3945         time_t timestamp;
3946         unsigned long flags;
3947         unsigned long nbuf;
3948         unsigned long nrec;
3949         char name[MAX_DB_NAME];
3950 };
3951
3952 struct backup_state {
3953         TALLOC_CTX *mem_ctx;
3954         struct ctdb_rec_buffer *recbuf;
3955         uint32_t db_id;
3956         int fd;
3957         unsigned int nbuf, nrec;
3958 };
3959
3960 static int backup_handler(uint32_t reqid, struct ctdb_ltdb_header *header,
3961                           TDB_DATA key, TDB_DATA data, void *private_data)
3962 {
3963         struct backup_state *state = (struct backup_state *)private_data;
3964         size_t len;
3965         int ret;
3966
3967         if (state->recbuf == NULL) {
3968                 state->recbuf = ctdb_rec_buffer_init(state->mem_ctx,
3969                                                      state->db_id);
3970                 if (state->recbuf == NULL) {
3971                         return ENOMEM;
3972                 }
3973         }
3974
3975         ret = ctdb_rec_buffer_add(state->recbuf, state->recbuf, reqid,
3976                                   header, key, data);
3977         if (ret != 0) {
3978                 return ret;
3979         }
3980
3981         len = ctdb_rec_buffer_len(state->recbuf);
3982         if (len < MAX_REC_BUFFER_SIZE) {
3983                 return 0;
3984         }
3985
3986         ret = ctdb_rec_buffer_write(state->recbuf, state->fd);
3987         if (ret != 0) {
3988                 fprintf(stderr, "Failed to write records to backup file\n");
3989                 return ret;
3990         }
3991
3992         state->nbuf += 1;
3993         state->nrec += state->recbuf->count;
3994         TALLOC_FREE(state->recbuf);
3995
3996         return 0;
3997 }
3998
3999 static int control_backupdb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4000                             int argc, const char **argv)
4001 {
4002         const char *db_name;
4003         struct ctdb_db_context *db;
4004         uint32_t db_id;
4005         uint8_t db_flags;
4006         struct backup_state state;
4007         struct db_header db_hdr;
4008         int fd, ret;
4009
4010         if (argc != 2) {
4011                 usage("backupdb");
4012         }
4013
4014         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
4015                 return 1;
4016         }
4017
4018         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4019                           db_flags, &db);
4020         if (ret != 0) {
4021                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4022                 return ret;
4023         }
4024
4025         fd = open(argv[1], O_RDWR|O_CREAT, 0600);
4026         if (fd == -1) {
4027                 ret = errno;
4028                 fprintf(stderr, "Failed to open file %s for writing\n",
4029                         argv[1]);
4030                 return ret;
4031         }
4032
4033         /* Write empty header first */
4034         ZERO_STRUCT(db_hdr);
4035         ret = write(fd, &db_hdr, sizeof(struct db_header));
4036         if (ret == -1) {
4037                 ret = errno;
4038                 close(fd);
4039                 fprintf(stderr, "Failed to write header to file %s\n", argv[1]);
4040                 return ret;
4041         }
4042
4043         state.mem_ctx = mem_ctx;
4044         state.recbuf = NULL;
4045         state.fd = fd;
4046         state.nbuf = 0;
4047         state.nrec = 0;
4048
4049         ret = ctdb_db_traverse_local(db, true, false, backup_handler, &state);
4050         if (ret != 0) {
4051                 fprintf(stderr, "Failed to collect records from DB %s\n",
4052                         db_name);
4053                 close(fd);
4054                 return ret;
4055         }
4056
4057         if (state.recbuf != NULL) {
4058                 ret = ctdb_rec_buffer_write(state.recbuf, state.fd);
4059                 if (ret != 0) {
4060                         fprintf(stderr,
4061                                 "Failed to write records to backup file\n");
4062                         close(fd);
4063                         return ret;
4064                 }
4065
4066                 state.nbuf += 1;
4067                 state.nrec += state.recbuf->count;
4068                 TALLOC_FREE(state.recbuf);
4069         }
4070
4071         db_hdr.version = DB_VERSION;
4072         db_hdr.timestamp = time(NULL);
4073         db_hdr.flags = db_flags;
4074         db_hdr.nbuf = state.nbuf;
4075         db_hdr.nrec = state.nrec;
4076         strncpy(db_hdr.name, db_name, MAX_DB_NAME-1);
4077
4078         lseek(fd, 0, SEEK_SET);
4079         ret = write(fd, &db_hdr, sizeof(struct db_header));
4080         if (ret == -1) {
4081                 ret = errno;
4082                 close(fd);
4083                 fprintf(stderr, "Failed to write header to file %s\n", argv[1]);
4084                 return ret;
4085         }
4086
4087         close(fd);
4088         printf("Database backed up to %s\n", argv[1]);
4089         return 0;
4090 }
4091
4092 static int control_restoredb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4093                              int argc, const char **argv)
4094 {
4095         const char *db_name = NULL;
4096         struct ctdb_db_context *db;
4097         struct db_header db_hdr;
4098         struct ctdb_node_map *nodemap;
4099         struct ctdb_req_control request;
4100         struct ctdb_reply_control **reply;
4101         struct ctdb_transdb wipedb;
4102         struct ctdb_pulldb_ext pulldb;
4103         struct ctdb_rec_buffer *recbuf;
4104         uint32_t generation;
4105         uint32_t *pnn_list;
4106         char timebuf[128];
4107         ssize_t n;
4108         int fd, i;
4109         int count, ret;
4110         uint8_t db_flags;
4111
4112         if (argc < 1 || argc > 2) {
4113                 usage("restoredb");
4114         }
4115
4116         fd = open(argv[0], O_RDONLY, 0600);
4117         if (fd == -1) {
4118                 ret = errno;
4119                 fprintf(stderr, "Failed to open file %s for reading\n",
4120                         argv[0]);
4121                 return ret;
4122         }
4123
4124         if (argc == 2) {
4125                 db_name = argv[1];
4126         }
4127
4128         n = read(fd, &db_hdr, sizeof(struct db_header));
4129         if (n == -1) {
4130                 ret = errno;
4131                 close(fd);
4132                 fprintf(stderr, "Failed to read db header from file %s\n",
4133                         argv[0]);
4134                 return ret;
4135         }
4136         db_hdr.name[sizeof(db_hdr.name)-1] = '\0';
4137
4138         if (db_hdr.version != DB_VERSION) {
4139                 fprintf(stderr,
4140                         "Wrong version of backup file, expected %u, got %lu\n",
4141                         DB_VERSION, db_hdr.version);
4142                 close(fd);
4143                 return EINVAL;
4144         }
4145
4146         if (db_name == NULL) {
4147                 db_name = db_hdr.name;
4148         }
4149
4150         strftime(timebuf, sizeof(timebuf)-1, "%Y/%m/%d %H:%M:%S",
4151                  localtime(&db_hdr.timestamp));
4152         printf("Restoring database %s from backup @ %s\n", db_name, timebuf);
4153
4154         db_flags = db_hdr.flags & 0xff;
4155         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4156                           db_flags, &db);
4157         if (ret != 0) {
4158                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4159                 close(fd);
4160                 return ret;
4161         }
4162
4163         nodemap = get_nodemap(ctdb, false);
4164         if (nodemap == NULL) {
4165                 fprintf(stderr, "Failed to get nodemap\n");
4166                 close(fd);
4167                 return ENOMEM;
4168         }
4169
4170         ret = get_generation(mem_ctx, ctdb, &generation);
4171         if (ret != 0) {
4172                 fprintf(stderr, "Failed to get current generation\n");
4173                 close(fd);
4174                 return ret;
4175         }
4176
4177         count = list_of_active_nodes(nodemap, CTDB_UNKNOWN_PNN, mem_ctx,
4178                                      &pnn_list);
4179         if (count <= 0) {
4180                 close(fd);
4181                 return ENOMEM;
4182         }
4183
4184         wipedb.db_id = ctdb_db_id(db);
4185         wipedb.tid = generation;
4186
4187         ctdb_req_control_db_freeze(&request, wipedb.db_id);
4188         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4189                                         ctdb->client, pnn_list, count,
4190                                         TIMEOUT(), &request, NULL, NULL);
4191         if (ret != 0) {
4192                 goto failed;
4193         }
4194
4195
4196         ctdb_req_control_db_transaction_start(&request, &wipedb);
4197         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4198                                         pnn_list, count, TIMEOUT(),
4199                                         &request, NULL, NULL);
4200         if (ret != 0) {
4201                 goto failed;
4202         }
4203
4204         ctdb_req_control_wipe_database(&request, &wipedb);
4205         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4206                                         pnn_list, count, TIMEOUT(),
4207                                         &request, NULL, NULL);
4208         if (ret != 0) {
4209                 goto failed;
4210         }
4211
4212         pulldb.db_id = ctdb_db_id(db);
4213         pulldb.lmaster = 0;
4214         pulldb.srvid = SRVID_CTDB_PUSHDB;
4215
4216         ctdb_req_control_db_push_start(&request, &pulldb);
4217         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4218                                         pnn_list, count, TIMEOUT(),
4219                                         &request, NULL, NULL);
4220         if (ret != 0) {
4221                 goto failed;
4222         }
4223
4224         for (i=0; i<db_hdr.nbuf; i++) {
4225                 struct ctdb_req_message message;
4226                 TDB_DATA data;
4227                 size_t np;
4228
4229                 ret = ctdb_rec_buffer_read(fd, mem_ctx, &recbuf);
4230                 if (ret != 0) {
4231                         goto failed;
4232                 }
4233
4234                 data.dsize = ctdb_rec_buffer_len(recbuf);
4235                 data.dptr = talloc_size(mem_ctx, data.dsize);
4236                 if (data.dptr == NULL) {
4237                         goto failed;
4238                 }
4239
4240                 ctdb_rec_buffer_push(recbuf, data.dptr, &np);
4241
4242                 message.srvid = pulldb.srvid;
4243                 message.data.data = data;
4244
4245                 ret = ctdb_client_message_multi(mem_ctx, ctdb->ev,
4246                                                 ctdb->client,
4247                                                 pnn_list, count,
4248                                                 &message, NULL);
4249                 if (ret != 0) {
4250                         goto failed;
4251                 }
4252
4253                 talloc_free(recbuf);
4254                 talloc_free(data.dptr);
4255         }
4256
4257         ctdb_req_control_db_push_confirm(&request, pulldb.db_id);
4258         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4259                                         pnn_list, count, TIMEOUT(),
4260                                         &request, NULL, &reply);
4261         if (ret != 0) {
4262                 goto failed;
4263         }
4264
4265         for (i=0; i<count; i++) {
4266                 uint32_t num_records;
4267
4268                 ret = ctdb_reply_control_db_push_confirm(reply[i],
4269                                                          &num_records);
4270                 if (ret != 0) {
4271                         fprintf(stderr, "Invalid response from node %u\n",
4272                                 pnn_list[i]);
4273                         goto failed;
4274                 }
4275
4276                 if (num_records != db_hdr.nrec) {
4277                         fprintf(stderr, "Node %u received %u of %lu records\n",
4278                                 pnn_list[i], num_records, db_hdr.nrec);
4279                         goto failed;
4280                 }
4281         }
4282
4283         ctdb_req_control_db_set_healthy(&request, wipedb.db_id);
4284         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4285                                         pnn_list, count, TIMEOUT(),
4286                                         &request, NULL, NULL);
4287         if (ret != 0) {
4288                 goto failed;
4289         }
4290
4291         ctdb_req_control_db_transaction_commit(&request, &wipedb);
4292         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4293                                         pnn_list, count, TIMEOUT(),
4294                                         &request, NULL, NULL);
4295         if (ret != 0) {
4296                 goto failed;
4297         }
4298
4299         ctdb_req_control_db_thaw(&request, wipedb.db_id);
4300         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4301                                         ctdb->client, pnn_list, count,
4302                                         TIMEOUT(), &request, NULL, NULL);
4303         if (ret != 0) {
4304                 goto failed;
4305         }
4306
4307         printf("Database %s restored\n", db_name);
4308         close(fd);
4309         return 0;
4310
4311
4312 failed:
4313         close(fd);
4314         ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
4315                               ctdb->pnn, TIMEOUT(), CTDB_RECOVERY_ACTIVE);
4316         return ret;
4317 }
4318
4319 struct dumpdbbackup_state {
4320         ctdb_rec_parser_func_t parser;
4321         struct dump_record_state sub_state;
4322 };
4323
4324 static int dumpdbbackup_handler(uint32_t reqid,
4325                                 struct ctdb_ltdb_header *header,
4326                                 TDB_DATA key, TDB_DATA data,
4327                                 void *private_data)
4328 {
4329         struct dumpdbbackup_state *state =
4330                 (struct dumpdbbackup_state *)private_data;
4331         struct ctdb_ltdb_header hdr;
4332         int ret;
4333
4334         ret = ctdb_ltdb_header_extract(&data, &hdr);
4335         if (ret != 0) {
4336                 return ret;
4337         }
4338
4339         return state->parser(reqid, &hdr, key, data, &state->sub_state);
4340 }
4341
4342 static int control_dumpdbbackup(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4343                                 int argc, const char **argv)
4344 {
4345         struct db_header db_hdr;
4346         char timebuf[128];
4347         struct dumpdbbackup_state state;
4348         ssize_t n;
4349         int fd, ret, i;
4350
4351         if (argc != 1) {
4352                 usage("dumpbackup");
4353         }
4354
4355         fd = open(argv[0], O_RDONLY, 0600);
4356         if (fd == -1) {
4357                 ret = errno;
4358                 fprintf(stderr, "Failed to open file %s for reading\n",
4359                         argv[0]);
4360                 return ret;
4361         }
4362
4363         n = read(fd, &db_hdr, sizeof(struct db_header));
4364         if (n == -1) {
4365                 ret = errno;
4366                 close(fd);
4367                 fprintf(stderr, "Failed to read db header from file %s\n",
4368                         argv[0]);
4369                 return ret;
4370         }
4371         db_hdr.name[sizeof(db_hdr.name)-1] = '\0';
4372
4373         if (db_hdr.version != DB_VERSION) {
4374                 fprintf(stderr,
4375                         "Wrong version of backup file, expected %u, got %lu\n",
4376                         DB_VERSION, db_hdr.version);
4377                 close(fd);
4378                 return EINVAL;
4379         }
4380
4381         strftime(timebuf, sizeof(timebuf)-1, "%Y/%m/%d %H:%M:%S",
4382                  localtime(&db_hdr.timestamp));
4383         printf("Dumping database %s from backup @ %s\n",
4384                db_hdr.name, timebuf);
4385
4386         state.parser = dump_record;
4387         state.sub_state.count = 0;
4388
4389         for (i=0; i<db_hdr.nbuf; i++) {
4390                 struct ctdb_rec_buffer *recbuf;
4391
4392                 ret = ctdb_rec_buffer_read(fd, mem_ctx, &recbuf);
4393                 if (ret != 0) {
4394                         fprintf(stderr, "Failed to read records\n");
4395                         close(fd);
4396                         return ret;
4397                 }
4398
4399                 ret = ctdb_rec_buffer_traverse(recbuf, dumpdbbackup_handler,
4400                                                &state);
4401                 if (ret != 0) {
4402                         fprintf(stderr, "Failed to dump records\n");
4403                         close(fd);
4404                         return ret;
4405                 }
4406         }
4407
4408         close(fd);
4409         printf("Dumped %u record(s)\n", state.sub_state.count);
4410         return 0;
4411 }
4412
4413 static int control_wipedb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4414                           int argc, const char **argv)
4415 {
4416         const char *db_name;
4417         struct ctdb_db_context *db;
4418         uint32_t db_id;
4419         uint8_t db_flags;
4420         struct ctdb_node_map *nodemap;
4421         struct ctdb_req_control request;
4422         struct ctdb_transdb wipedb;
4423         uint32_t generation;
4424         uint32_t *pnn_list;
4425         int count, ret;
4426
4427         if (argc != 1) {
4428                 usage("wipedb");
4429         }
4430
4431         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
4432                 return 1;
4433         }
4434
4435         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4436                           db_flags, &db);
4437         if (ret != 0) {
4438                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4439                 return ret;
4440         }
4441
4442         nodemap = get_nodemap(ctdb, false);
4443         if (nodemap == NULL) {
4444                 fprintf(stderr, "Failed to get nodemap\n");
4445                 return ENOMEM;
4446         }
4447
4448         ret = get_generation(mem_ctx, ctdb, &generation);
4449         if (ret != 0) {
4450                 fprintf(stderr, "Failed to get current generation\n");
4451                 return ret;
4452         }
4453
4454         count = list_of_active_nodes(nodemap, CTDB_UNKNOWN_PNN, mem_ctx,
4455                                      &pnn_list);
4456         if (count <= 0) {
4457                 return ENOMEM;
4458         }
4459
4460         ctdb_req_control_db_freeze(&request, db_id);
4461         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4462                                         ctdb->client, pnn_list, count,
4463                                         TIMEOUT(), &request, NULL, NULL);
4464         if (ret != 0) {
4465                 goto failed;
4466         }
4467
4468         wipedb.db_id = db_id;
4469         wipedb.tid = generation;
4470
4471         ctdb_req_control_db_transaction_start(&request, &wipedb);
4472         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4473                                         pnn_list, count, TIMEOUT(),
4474                                         &request, NULL, NULL);
4475         if (ret != 0) {
4476                 goto failed;
4477         }
4478
4479         ctdb_req_control_wipe_database(&request, &wipedb);
4480         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4481                                         pnn_list, count, TIMEOUT(),
4482                                         &request, NULL, NULL);
4483         if (ret != 0) {
4484                 goto failed;
4485         }
4486
4487         ctdb_req_control_db_set_healthy(&request, db_id);
4488         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4489                                         pnn_list, count, TIMEOUT(),
4490                                         &request, NULL, NULL);
4491         if (ret != 0) {
4492                 goto failed;
4493         }
4494
4495         ctdb_req_control_db_transaction_commit(&request, &wipedb);
4496         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4497                                         pnn_list, count, TIMEOUT(),
4498                                         &request, NULL, NULL);
4499         if (ret != 0) {
4500                 goto failed;
4501         }
4502
4503         ctdb_req_control_db_thaw(&request, db_id);
4504         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4505                                         ctdb->client, pnn_list, count,
4506                                         TIMEOUT(), &request, NULL, NULL);
4507         if (ret != 0) {
4508                 goto failed;
4509         }
4510
4511         printf("Database %s wiped\n", db_name);
4512         return 0;
4513
4514
4515 failed:
4516         ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
4517                               ctdb->pnn, TIMEOUT(), CTDB_RECOVERY_ACTIVE);
4518         return ret;
4519 }
4520
4521 static int control_recmaster(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4522                              int argc, const char **argv)
4523 {
4524         uint32_t recmaster;
4525         int ret;
4526
4527         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
4528                                       ctdb->cmd_pnn, TIMEOUT(), &recmaster);
4529         if (ret != 0) {
4530                 return ret;
4531         }
4532
4533         printf("%u\n", recmaster);
4534         return 0;
4535 }
4536
4537 static int control_event(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4538                          int argc, const char **argv)
4539 {
4540         char *t, *event_helper = NULL;
4541         char *eventd_socket = NULL;
4542         const char **new_argv;
4543         int i;
4544
4545         t = getenv("CTDB_EVENT_HELPER");
4546         if (t != NULL) {
4547                 event_helper = talloc_strdup(mem_ctx, t);
4548         } else {
4549                 event_helper = talloc_asprintf(mem_ctx, "%s/ctdb_event",
4550                                                CTDB_HELPER_BINDIR);
4551         }
4552
4553         if (event_helper == NULL) {
4554                 fprintf(stderr, "Unable to set event daemon helper\n");
4555                 return 1;
4556         }
4557
4558         t = getenv("CTDB_SOCKET");
4559         if (t != NULL) {
4560                 eventd_socket = talloc_asprintf(mem_ctx, "%s/eventd.sock",
4561                                                 dirname(t));
4562         } else {
4563                 eventd_socket = talloc_asprintf(mem_ctx, "%s/eventd.sock",
4564                                                 CTDB_RUNDIR);
4565         }
4566
4567         if (eventd_socket == NULL) {
4568                 fprintf(stderr, "Unable to set event daemon socket\n");
4569                 return 1;
4570         }
4571
4572         new_argv = talloc_array(mem_ctx, const char *, argc + 1);
4573         if (new_argv == NULL) {
4574                 fprintf(stderr, "Memory allocation error\n");
4575                 return 1;
4576         }
4577
4578         new_argv[0] = eventd_socket;
4579         for (i=0; i<argc; i++) {
4580                 new_argv[i+1] = argv[i];
4581         }
4582
4583         return run_helper(mem_ctx, "event daemon helper", event_helper,
4584                           argc+1, new_argv);
4585 }
4586
4587 static int control_scriptstatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4588                                 int argc, const char **argv)
4589 {
4590         const char *new_argv[3];
4591
4592         if (argc > 1) {
4593                 usage("scriptstatus");
4594         }
4595
4596         new_argv[0] = "status";
4597         new_argv[1] = (argc == 0) ? "monitor" : argv[0];
4598         new_argv[2] = NULL;
4599
4600         (void) control_event(mem_ctx, ctdb, 2, new_argv);
4601         return 0;
4602 }
4603
4604 static int control_natgw(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4605                          int argc, const char **argv)
4606 {
4607         char *t, *natgw_helper = NULL;
4608
4609         if (argc != 1) {
4610                 usage("natgw");
4611         }
4612
4613         t = getenv("CTDB_NATGW_HELPER");
4614         if (t != NULL) {
4615                 natgw_helper = talloc_strdup(mem_ctx, t);
4616         } else {
4617                 natgw_helper = talloc_asprintf(mem_ctx, "%s/ctdb_natgw",
4618                                                CTDB_HELPER_BINDIR);
4619         }
4620
4621         if (natgw_helper == NULL) {
4622                 fprintf(stderr, "Unable to set NAT gateway helper\n");
4623                 return 1;
4624         }
4625
4626         return run_helper(mem_ctx, "NAT gateway helper", natgw_helper,
4627                           argc, argv);
4628 }
4629
4630 /*
4631  * Find the PNN of the current node
4632  * discover the pnn by loading the nodes file and try to bind
4633  * to all addresses one at a time until the ip address is found.
4634  */
4635 static bool find_node_xpnn(TALLOC_CTX *mem_ctx, uint32_t *pnn)
4636 {
4637         struct ctdb_node_map *nodemap;
4638         int i;
4639
4640         nodemap = read_nodes_file(mem_ctx, CTDB_UNKNOWN_PNN);
4641         if (nodemap == NULL) {
4642                 return false;
4643         }
4644
4645         for (i=0; i<nodemap->num; i++) {
4646                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
4647                         continue;
4648                 }
4649                 if (ctdb_sys_have_ip(&nodemap->node[i].addr)) {
4650                         if (pnn != NULL) {
4651                                 *pnn = nodemap->node[i].pnn;
4652                         }
4653                         talloc_free(nodemap);
4654                         return true;
4655                 }
4656         }
4657
4658         fprintf(stderr, "Failed to detect PNN of the current node.\n");
4659         talloc_free(nodemap);
4660         return false;
4661 }
4662
4663 static int control_getreclock(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4664                               int argc, const char **argv)
4665 {
4666         const char *reclock;
4667         int ret;
4668
4669         if (argc != 0) {
4670                 usage("getreclock");
4671         }
4672
4673         ret = ctdb_ctrl_get_reclock_file(mem_ctx, ctdb->ev, ctdb->client,
4674                                          ctdb->cmd_pnn, TIMEOUT(), &reclock);
4675         if (ret != 0) {
4676                 return ret;
4677         }
4678
4679         if (reclock != NULL) {
4680                 printf("%s\n", reclock);
4681         }
4682
4683         return 0;
4684 }
4685
4686 static int control_setlmasterrole(TALLOC_CTX *mem_ctx,
4687                                   struct ctdb_context *ctdb,
4688                                   int argc, const char **argv)
4689 {
4690         uint32_t lmasterrole = 0;
4691         int ret;
4692
4693         if (argc != 1) {
4694                 usage("setlmasterrole");
4695         }
4696
4697         if (strcmp(argv[0], "on") == 0) {
4698                 lmasterrole = 1;
4699         } else if (strcmp(argv[0], "off") == 0) {
4700                 lmasterrole = 0;
4701         } else {
4702                 usage("setlmasterrole");
4703         }
4704
4705         ret = ctdb_ctrl_set_lmasterrole(mem_ctx, ctdb->ev, ctdb->client,
4706                                         ctdb->cmd_pnn, TIMEOUT(), lmasterrole);
4707         if (ret != 0) {
4708                 return ret;
4709         }
4710
4711         return 0;
4712 }
4713
4714 static int control_setrecmasterrole(TALLOC_CTX *mem_ctx,
4715                                     struct ctdb_context *ctdb,
4716                                     int argc, const char **argv)
4717 {
4718         uint32_t recmasterrole = 0;
4719         int ret;
4720
4721         if (argc != 1) {
4722                 usage("setrecmasterrole");
4723         }
4724
4725         if (strcmp(argv[0], "on") == 0) {
4726                 recmasterrole = 1;
4727         } else if (strcmp(argv[0], "off") == 0) {
4728                 recmasterrole = 0;
4729         } else {
4730                 usage("setrecmasterrole");
4731         }
4732
4733         ret = ctdb_ctrl_set_recmasterrole(mem_ctx, ctdb->ev, ctdb->client,
4734                                           ctdb->cmd_pnn, TIMEOUT(),
4735                                           recmasterrole);
4736         if (ret != 0) {
4737                 return ret;
4738         }
4739
4740         return 0;
4741 }
4742
4743 static int control_setdbreadonly(TALLOC_CTX *mem_ctx,
4744                                  struct ctdb_context *ctdb,
4745                                  int argc, const char **argv)
4746 {
4747         uint32_t db_id;
4748         uint8_t db_flags;
4749         int ret;
4750
4751         if (argc != 1) {
4752                 usage("setdbreadonly");
4753         }
4754
4755         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, NULL, &db_flags)) {
4756                 return 1;
4757         }
4758
4759         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
4760                 fprintf(stderr, "READONLY can be set only on volatile DB\n");
4761                 return 1;
4762         }
4763
4764         ret = ctdb_ctrl_set_db_readonly(mem_ctx, ctdb->ev, ctdb->client,
4765                                         ctdb->cmd_pnn, TIMEOUT(), db_id);
4766         if (ret != 0) {
4767                 return ret;
4768         }
4769
4770         return 0;
4771 }
4772
4773 static int control_setdbsticky(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4774                                int argc, const char **argv)
4775 {
4776         uint32_t db_id;
4777         uint8_t db_flags;
4778         int ret;
4779
4780         if (argc != 1) {
4781                 usage("setdbsticky");
4782         }
4783
4784         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, NULL, &db_flags)) {
4785                 return 1;
4786         }
4787
4788         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
4789                 fprintf(stderr, "STICKY can be set only on volatile DB\n");
4790                 return 1;
4791         }
4792
4793         ret = ctdb_ctrl_set_db_sticky(mem_ctx, ctdb->ev, ctdb->client,
4794                                       ctdb->cmd_pnn, TIMEOUT(), db_id);
4795         if (ret != 0) {
4796                 return ret;
4797         }
4798
4799         return 0;
4800 }
4801
4802 static int control_pfetch(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4803                           int argc, const char **argv)
4804 {
4805         const char *db_name;
4806         struct ctdb_db_context *db;
4807         struct ctdb_transaction_handle *h;
4808         uint8_t db_flags;
4809         TDB_DATA key, data;
4810         int ret;
4811
4812         if (argc < 2 || argc > 3) {
4813                 usage("pfetch");
4814         }
4815
4816         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
4817                 return 1;
4818         }
4819
4820         if (! (db_flags &
4821                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
4822                 fprintf(stderr, "Transactions not supported on DB %s\n",
4823                         db_name);
4824                 return 1;
4825         }
4826
4827         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4828                           db_flags, &db);
4829         if (ret != 0) {
4830                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4831                 return ret;
4832         }
4833
4834         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
4835         if (ret != 0) {
4836                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
4837                 return ret;
4838         }
4839
4840         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
4841                                      TIMEOUT(), db, true, &h);
4842         if (ret != 0) {
4843                 fprintf(stderr, "Failed to start transaction on db %s\n",
4844                         db_name);
4845                 return ret;
4846         }
4847
4848         ret = ctdb_transaction_fetch_record(h, key, mem_ctx, &data);
4849         if (ret != 0) {
4850                 fprintf(stderr, "Failed to read record for key %s\n",
4851                         argv[1]);
4852                 ctdb_transaction_cancel(h);
4853                 return ret;
4854         }
4855
4856         printf("%.*s\n", (int)data.dsize, data.dptr);
4857
4858         ctdb_transaction_cancel(h);
4859         return 0;
4860 }
4861
4862 static int control_pstore(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4863                           int argc, const char **argv)
4864 {
4865         const char *db_name;
4866         struct ctdb_db_context *db;
4867         struct ctdb_transaction_handle *h;
4868         uint8_t db_flags;
4869         TDB_DATA key, data;
4870         int ret;
4871
4872         if (argc != 3) {
4873                 usage("pstore");
4874         }
4875
4876         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
4877                 return 1;
4878         }
4879
4880         if (! (db_flags &
4881                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
4882                 fprintf(stderr, "Transactions not supported on DB %s\n",
4883                         db_name);
4884                 return 1;
4885         }
4886
4887         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4888                           db_flags, &db);
4889         if (ret != 0) {
4890                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4891                 return ret;
4892         }
4893
4894         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
4895         if (ret != 0) {
4896                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
4897                 return ret;
4898         }
4899
4900         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &data);
4901         if (ret != 0) {
4902                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
4903                 return ret;
4904         }
4905
4906         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
4907                                      TIMEOUT(), db, false, &h);
4908         if (ret != 0) {
4909                 fprintf(stderr, "Failed to start transaction on db %s\n",
4910                         db_name);
4911                 return ret;
4912         }
4913
4914         ret = ctdb_transaction_store_record(h, key, data);
4915         if (ret != 0) {
4916                 fprintf(stderr, "Failed to store record for key %s\n",
4917                         argv[1]);
4918                 ctdb_transaction_cancel(h);
4919                 return ret;
4920         }
4921
4922         ret = ctdb_transaction_commit(h);
4923         if (ret != 0) {
4924                 fprintf(stderr, "Failed to commit transaction on db %s\n",
4925                         db_name);
4926                 ctdb_transaction_cancel(h);
4927                 return ret;
4928         }
4929
4930         return 0;
4931 }
4932
4933 static int control_pdelete(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4934                            int argc, const char **argv)
4935 {
4936         const char *db_name;
4937         struct ctdb_db_context *db;
4938         struct ctdb_transaction_handle *h;
4939         uint8_t db_flags;
4940         TDB_DATA key;
4941         int ret;
4942
4943         if (argc != 2) {
4944                 usage("pdelete");
4945         }
4946
4947         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
4948                 return 1;
4949         }
4950
4951         if (! (db_flags &
4952                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
4953                 fprintf(stderr, "Transactions not supported on DB %s\n",
4954                         db_name);
4955                 return 1;
4956         }
4957
4958         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4959                           db_flags, &db);
4960         if (ret != 0) {
4961                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4962                 return ret;
4963         }
4964
4965         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
4966         if (ret != 0) {
4967                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
4968                 return ret;
4969         }
4970
4971         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
4972                                      TIMEOUT(), db, false, &h);
4973         if (ret != 0) {
4974                 fprintf(stderr, "Failed to start transaction on db %s\n",
4975                         db_name);
4976                 return ret;
4977         }
4978
4979         ret = ctdb_transaction_delete_record(h, key);
4980         if (ret != 0) {
4981                 fprintf(stderr, "Failed to delete record for key %s\n",
4982                         argv[1]);
4983                 ctdb_transaction_cancel(h);
4984                 return ret;
4985         }
4986
4987         ret = ctdb_transaction_commit(h);
4988         if (ret != 0) {
4989                 fprintf(stderr, "Failed to commit transaction on db %s\n",
4990                         db_name);
4991                 ctdb_transaction_cancel(h);
4992                 return ret;
4993         }
4994
4995         return 0;
4996 }
4997
4998 static int ptrans_parse_string(TALLOC_CTX *mem_ctx, const char **ptr, TDB_DATA *data)
4999 {
5000         const char *t;
5001         size_t n;
5002         int ret;
5003
5004         *data = tdb_null;
5005
5006         /* Skip whitespace */
5007         n = strspn(*ptr, " \t");
5008         t = *ptr + n;
5009
5010         if (t[0] == '"') {
5011                 /* Quoted ASCII string - no wide characters! */
5012                 t++;
5013                 n = strcspn(t, "\"");
5014                 if (t[n] == '"') {
5015                         if (n > 0) {
5016                                 ret = str_to_data(t, n, mem_ctx, data);
5017                                 if (ret != 0) {
5018                                         return ret;
5019                                 }
5020                         }
5021                         *ptr = t + n + 1;
5022                 } else {
5023                         fprintf(stderr, "Unmatched \" in input %s\n", *ptr);
5024                         return 1;
5025                 }
5026         } else {
5027                 fprintf(stderr, "Unsupported input format in %s\n", *ptr);
5028                 return 1;
5029         }
5030
5031         return 0;
5032 }
5033
5034 #define MAX_LINE_SIZE   1024
5035
5036 static bool ptrans_get_key_value(TALLOC_CTX *mem_ctx, FILE *file,
5037                                  TDB_DATA *key, TDB_DATA *value)
5038 {
5039         char line [MAX_LINE_SIZE]; /* FIXME: make this more flexible? */
5040         const char *ptr;
5041         int ret;
5042
5043         ptr = fgets(line, MAX_LINE_SIZE, file);
5044         if (ptr == NULL) {
5045                 return false;
5046         }
5047
5048         /* Get key */
5049         ret = ptrans_parse_string(mem_ctx, &ptr, key);
5050         if (ret != 0 || ptr == NULL || key->dptr == NULL) {
5051                 /* Line Ignored but not EOF */
5052                 *key = tdb_null;
5053                 return true;
5054         }
5055
5056         /* Get value */
5057         ret = ptrans_parse_string(mem_ctx, &ptr, value);
5058         if (ret != 0) {
5059                 /* Line Ignored but not EOF */
5060                 talloc_free(key->dptr);
5061                 *key = tdb_null;
5062                 return true;
5063         }
5064
5065         return true;
5066 }
5067
5068 static int control_ptrans(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5069                           int argc, const char **argv)
5070 {
5071         const char *db_name;
5072         struct ctdb_db_context *db;
5073         struct ctdb_transaction_handle *h;
5074         uint8_t db_flags;
5075         FILE *file;
5076         TDB_DATA key, value;
5077         int ret;
5078
5079         if (argc < 1 || argc > 2) {
5080                 usage("ptrans");
5081         }
5082
5083         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5084                 return 1;
5085         }
5086
5087         if (! (db_flags &
5088                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
5089                 fprintf(stderr, "Transactions not supported on DB %s\n",
5090                         db_name);
5091                 return 1;
5092         }
5093
5094         if (argc == 2) {
5095                 file = fopen(argv[1], "r");
5096                 if (file == NULL) {
5097                         fprintf(stderr, "Failed to open file %s\n", argv[1]);
5098                         return 1;
5099                 }
5100         } else {
5101                 file = stdin;
5102         }
5103
5104         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5105                           db_flags, &db);
5106         if (ret != 0) {
5107                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5108                 goto done;
5109         }
5110
5111         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
5112                                      TIMEOUT(), db, false, &h);
5113         if (ret != 0) {
5114                 fprintf(stderr, "Failed to start transaction on db %s\n",
5115                         db_name);
5116                 goto done;
5117         }
5118
5119         while (ptrans_get_key_value(mem_ctx, file, &key, &value)) {
5120                 if (key.dsize != 0) {
5121                         ret = ctdb_transaction_store_record(h, key, value);
5122                         if (ret != 0) {
5123                                 fprintf(stderr, "Failed to store record\n");
5124                                 ctdb_transaction_cancel(h);
5125                                 goto done;
5126                         }
5127                         talloc_free(key.dptr);
5128                         talloc_free(value.dptr);
5129                 }
5130         }
5131
5132         ret = ctdb_transaction_commit(h);
5133         if (ret != 0) {
5134                 fprintf(stderr, "Failed to commit transaction on db %s\n",
5135                         db_name);
5136                 ctdb_transaction_cancel(h);
5137         }
5138
5139 done:
5140         if (file != stdin) {
5141                 fclose(file);
5142         }
5143         return ret;
5144 }
5145
5146 static int control_tfetch(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5147                           int argc, const char **argv)
5148 {
5149         struct tdb_context *tdb;
5150         TDB_DATA key, data;
5151         struct ctdb_ltdb_header header;
5152         int ret;
5153
5154         if (argc < 2 || argc > 3) {
5155                 usage("tfetch");
5156         }
5157
5158         tdb = tdb_open(argv[0], 0, 0, O_RDWR, 0);
5159         if (tdb == NULL) {
5160                 fprintf(stderr, "Failed to open TDB file %s\n", argv[0]);
5161                 return 1;
5162         }
5163
5164         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5165         if (ret != 0) {
5166                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5167                 tdb_close(tdb);
5168                 return ret;
5169         }
5170
5171         data = tdb_fetch(tdb, key);
5172         if (data.dptr == NULL) {
5173                 fprintf(stderr, "No record for key %s\n", argv[1]);
5174                 tdb_close(tdb);
5175                 return 1;
5176         }
5177
5178         if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
5179                 fprintf(stderr, "Invalid record for key %s\n", argv[1]);
5180                 tdb_close(tdb);
5181                 return 1;
5182         }
5183
5184         tdb_close(tdb);
5185
5186         if (argc == 3) {
5187                 int fd;
5188                 ssize_t nwritten;
5189
5190                 fd = open(argv[2], O_WRONLY|O_CREAT|O_TRUNC, 0600);
5191                 if (fd == -1) {
5192                         fprintf(stderr, "Failed to open output file %s\n",
5193                                 argv[2]);
5194                         goto fail;
5195                 }
5196
5197                 nwritten = sys_write(fd, data.dptr, data.dsize);
5198                 if (nwritten != data.dsize) {
5199                         fprintf(stderr, "Failed to write record to file\n");
5200                         close(fd);
5201                         goto fail;
5202                 }
5203
5204                 close(fd);
5205         }
5206
5207 fail:
5208         ret = ctdb_ltdb_header_extract(&data, &header);
5209         if (ret != 0) {
5210                 fprintf(stderr, "Failed to parse header from data\n");
5211                 return 1;
5212         }
5213
5214         dump_ltdb_header(&header);
5215         dump_tdb_data("data", data);
5216
5217         return 0;
5218 }
5219
5220 static int control_tstore(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5221                           int argc, const char **argv)
5222 {
5223         struct tdb_context *tdb;
5224         TDB_DATA key, data[2], value;
5225         struct ctdb_ltdb_header header;
5226         uint8_t header_buf[sizeof(struct ctdb_ltdb_header)];
5227         size_t np;
5228         int ret;
5229
5230         if (argc < 3 || argc > 5) {
5231                 usage("tstore");
5232         }
5233
5234         tdb = tdb_open(argv[0], 0, 0, O_RDWR, 0);
5235         if (tdb == NULL) {
5236                 fprintf(stderr, "Failed to open TDB file %s\n", argv[0]);
5237                 return 1;
5238         }
5239
5240         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5241         if (ret != 0) {
5242                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5243                 tdb_close(tdb);
5244                 return ret;
5245         }
5246
5247         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &value);
5248         if (ret != 0) {
5249                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
5250                 tdb_close(tdb);
5251                 return ret;
5252         }
5253
5254         ZERO_STRUCT(header);
5255
5256         if (argc > 3) {
5257                 header.rsn = (uint64_t)strtoull(argv[3], NULL, 0);
5258         }
5259         if (argc > 4) {
5260                 header.dmaster = (uint32_t)atol(argv[4]);
5261         }
5262         if (argc > 5) {
5263                 header.flags = (uint32_t)atol(argv[5]);
5264         }
5265
5266         ctdb_ltdb_header_push(&header, header_buf, &np);
5267
5268         data[0].dsize = np;
5269         data[0].dptr = header_buf;
5270
5271         data[1].dsize = value.dsize;
5272         data[1].dptr = value.dptr;
5273
5274         ret = tdb_storev(tdb, key, data, 2, TDB_REPLACE);
5275         if (ret != 0) {
5276                 fprintf(stderr, "Failed to write record %s to file %s\n",
5277                         argv[1], argv[0]);
5278         }
5279
5280         tdb_close(tdb);
5281
5282         return ret;
5283 }
5284
5285 static int control_readkey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5286                            int argc, const char **argv)
5287 {
5288         const char *db_name;
5289         struct ctdb_db_context *db;
5290         struct ctdb_record_handle *h;
5291         uint8_t db_flags;
5292         TDB_DATA key, data;
5293         bool readonly = false;
5294         int ret;
5295
5296         if (argc < 2 || argc > 3) {
5297                 usage("readkey");
5298         }
5299
5300         if (argc == 3) {
5301                 if (strcmp(argv[2], "readonly") == 0) {
5302                         readonly = true;
5303                 } else {
5304                         usage("readkey");
5305                 }
5306         }
5307
5308         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5309                 return 1;
5310         }
5311
5312         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
5313                 fprintf(stderr, "DB %s is not a volatile database\n",
5314                         db_name);
5315                 return 1;
5316         }
5317
5318         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5319                           db_flags, &db);
5320         if (ret != 0) {
5321                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5322                 return ret;
5323         }
5324
5325         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5326         if (ret != 0) {
5327                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5328                 return ret;
5329         }
5330
5331         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5332                               db, key, readonly, &h, NULL, &data);
5333         if (ret != 0) {
5334                 fprintf(stderr, "Failed to read record for key %s\n",
5335                         argv[1]);
5336         } else {
5337                 printf("Data: size:%zu ptr:[%.*s]\n", data.dsize,
5338                        (int)data.dsize, data.dptr);
5339         }
5340
5341         talloc_free(h);
5342         return ret;
5343 }
5344
5345 static int control_writekey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5346                             int argc, const char **argv)
5347 {
5348         const char *db_name;
5349         struct ctdb_db_context *db;
5350         struct ctdb_record_handle *h;
5351         uint8_t db_flags;
5352         TDB_DATA key, data;
5353         int ret;
5354
5355         if (argc != 3) {
5356                 usage("writekey");
5357         }
5358
5359         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5360                 return 1;
5361         }
5362
5363         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
5364                 fprintf(stderr, "DB %s is not a volatile database\n",
5365                         db_name);
5366                 return 1;
5367         }
5368
5369         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5370                           db_flags, &db);
5371         if (ret != 0) {
5372                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5373                 return ret;
5374         }
5375
5376         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5377         if (ret != 0) {
5378                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5379                 return ret;
5380         }
5381
5382         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &data);
5383         if (ret != 0) {
5384                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
5385                 return ret;
5386         }
5387
5388         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5389                               db, key, false, &h, NULL, NULL);
5390         if (ret != 0) {
5391                 fprintf(stderr, "Failed to lock record for key %s\n", argv[0]);
5392                 return ret;
5393         }
5394
5395         ret = ctdb_store_record(h, data);
5396         if (ret != 0) {
5397                 fprintf(stderr, "Failed to store record for key %s\n",
5398                         argv[1]);
5399         }
5400
5401         talloc_free(h);
5402         return ret;
5403 }
5404
5405 static int control_deletekey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5406                              int argc, const char **argv)
5407 {
5408         const char *db_name;
5409         struct ctdb_db_context *db;
5410         struct ctdb_record_handle *h;
5411         uint8_t db_flags;
5412         TDB_DATA key, data;
5413         int ret;
5414
5415         if (argc != 2) {
5416                 usage("deletekey");
5417         }
5418
5419         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5420                 return 1;
5421         }
5422
5423         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
5424                 fprintf(stderr, "DB %s is not a volatile database\n",
5425                         db_name);
5426                 return 1;
5427         }
5428
5429         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5430                           db_flags, &db);
5431         if (ret != 0) {
5432                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5433                 return ret;
5434         }
5435
5436         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5437         if (ret != 0) {
5438                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5439                 return ret;
5440         }
5441
5442         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5443                               db, key, false, &h, NULL, &data);
5444         if (ret != 0) {
5445                 fprintf(stderr, "Failed to fetch record for key %s\n",
5446                         argv[1]);
5447                 return ret;
5448         }
5449
5450         ret = ctdb_delete_record(h);
5451         if (ret != 0) {
5452                 fprintf(stderr, "Failed to delete record for key %s\n",
5453                         argv[1]);
5454         }
5455
5456         talloc_free(h);
5457         return ret;
5458 }
5459
5460 static int control_checktcpport(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5461                                 int argc, const char **argv)
5462 {
5463         struct sockaddr_in sin;
5464         unsigned int port;
5465         int s, v;
5466         int ret;
5467
5468         if (argc != 1) {
5469                 usage("chktcpport");
5470         }
5471
5472         port = atoi(argv[0]);
5473
5474         s = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
5475         if (s == -1) {
5476                 fprintf(stderr, "Failed to open local socket\n");
5477                 return errno;
5478         }
5479
5480         v = fcntl(s, F_GETFL, 0);
5481         if (v == -1 || fcntl(s, F_SETFL, v | O_NONBLOCK)) {
5482                 fprintf(stderr, "Unable to set socket non-blocking\n");
5483                 close(s);
5484                 return errno;
5485         }
5486
5487         bzero(&sin, sizeof(sin));
5488         sin.sin_family = AF_INET;
5489         sin.sin_port = htons(port);
5490         ret = bind(s, (struct sockaddr *)&sin, sizeof(sin));
5491         close(s);
5492         if (ret == -1) {
5493                 fprintf(stderr, "Failed to bind to TCP port %u\n", port);
5494                 return errno;
5495         }
5496
5497         return 0;
5498 }
5499
5500 static int control_getdbseqnum(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5501                                int argc, const char **argv)
5502 {
5503         uint32_t db_id;
5504         const char *db_name;
5505         uint64_t seqnum;
5506         int ret;
5507
5508         if (argc != 1) {
5509                 usage("getdbseqnum");
5510         }
5511
5512         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, NULL)) {
5513                 return 1;
5514         }
5515
5516         ret = ctdb_ctrl_get_db_seqnum(mem_ctx, ctdb->ev, ctdb->client,
5517                                       ctdb->cmd_pnn, TIMEOUT(), db_id,
5518                                       &seqnum);
5519         if (ret != 0) {
5520                 fprintf(stderr, "Failed to get sequence number for DB %s\n",
5521                         db_name);
5522                 return ret;
5523         }
5524
5525         printf("0x%"PRIx64"\n", seqnum);
5526         return 0;
5527 }
5528
5529 static int control_nodestatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5530                               int argc, const char **argv)
5531 {
5532         const char *nodestring = NULL;
5533         struct ctdb_node_map *nodemap;
5534         int ret, i;
5535         bool print_hdr = false;
5536
5537         if (argc > 1) {
5538                 usage("nodestatus");
5539         }
5540
5541         if (argc == 1) {
5542                 nodestring = argv[0];
5543                 if (strcmp(nodestring, "all") == 0) {
5544                         print_hdr = true;
5545                 }
5546         }
5547
5548         if (! parse_nodestring(mem_ctx, ctdb, nodestring, &nodemap)) {
5549                 return 1;
5550         }
5551
5552         if (options.machinereadable) {
5553                 print_nodemap_machine(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn);
5554         } else {
5555                 print_nodemap(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn, print_hdr);
5556         }
5557
5558         ret = 0;
5559         for (i=0; i<nodemap->num; i++) {
5560                 ret |= nodemap->node[i].flags;
5561         }
5562
5563         return ret;
5564 }
5565
5566 const struct {
5567         const char *name;
5568         uint32_t offset;
5569 } db_stats_fields[] = {
5570 #define DBSTATISTICS_FIELD(n) { #n, offsetof(struct ctdb_db_statistics, n) }
5571         DBSTATISTICS_FIELD(db_ro_delegations),
5572         DBSTATISTICS_FIELD(db_ro_revokes),
5573         DBSTATISTICS_FIELD(locks.num_calls),
5574         DBSTATISTICS_FIELD(locks.num_current),
5575         DBSTATISTICS_FIELD(locks.num_pending),
5576         DBSTATISTICS_FIELD(locks.num_failed),
5577         DBSTATISTICS_FIELD(db_ro_delegations),
5578 };
5579
5580 static void print_dbstatistics(const char *db_name,
5581                                struct ctdb_db_statistics *s)
5582 {
5583         int i;
5584         const char *prefix = NULL;
5585         int preflen = 0;
5586
5587         printf("DB Statistics %s\n", db_name);
5588
5589         for (i=0; i<ARRAY_SIZE(db_stats_fields); i++) {
5590                 if (strchr(db_stats_fields[i].name, '.') != NULL) {
5591                         preflen = strcspn(db_stats_fields[i].name, ".") + 1;
5592                         if (! prefix ||
5593                             strncmp(prefix, db_stats_fields[i].name, preflen) != 0) {
5594                                 prefix = db_stats_fields[i].name;
5595                                 printf(" %*.*s\n", preflen-1, preflen-1,
5596                                        db_stats_fields[i].name);
5597                         }
5598                 } else {
5599                         preflen = 0;
5600                 }
5601                 printf(" %*s%-22s%*s%10u\n", preflen ? 4 : 0, "",
5602                        db_stats_fields[i].name+preflen, preflen ? 0 : 4, "",
5603                        *(uint32_t *)(db_stats_fields[i].offset+(uint8_t *)s));
5604         }
5605
5606         printf(" hop_count_buckets:");
5607         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
5608                 printf(" %d", s->hop_count_bucket[i]);
5609         }
5610         printf("\n");
5611
5612         printf(" lock_buckets:");
5613         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
5614                 printf(" %d", s->locks.buckets[i]);
5615         }
5616         printf("\n");
5617
5618         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
5619                "locks_latency      MIN/AVG/MAX",
5620                s->locks.latency.min, LATENCY_AVG(s->locks.latency),
5621                s->locks.latency.max, s->locks.latency.num);
5622
5623         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
5624                "vacuum_latency     MIN/AVG/MAX",
5625                s->vacuum.latency.min, LATENCY_AVG(s->vacuum.latency),
5626                s->vacuum.latency.max, s->vacuum.latency.num);
5627
5628         printf(" Num Hot Keys:     %d\n", s->num_hot_keys);
5629         for (i=0; i<s->num_hot_keys; i++) {
5630                 int j;
5631                 printf("     Count:%d Key:", s->hot_keys[i].count);
5632                 for (j=0; j<s->hot_keys[i].key.dsize; j++) {
5633                         printf("%02x", s->hot_keys[i].key.dptr[j] & 0xff);
5634                 }
5635                 printf("\n");
5636         }
5637 }
5638
5639 static int control_dbstatistics(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5640                                 int argc, const char **argv)
5641 {
5642         uint32_t db_id;
5643         const char *db_name;
5644         struct ctdb_db_statistics *dbstats;
5645         int ret;
5646
5647         if (argc != 1) {
5648                 usage("dbstatistics");
5649         }
5650
5651         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, NULL)) {
5652                 return 1;
5653         }
5654
5655         ret = ctdb_ctrl_get_db_statistics(mem_ctx, ctdb->ev, ctdb->client,
5656                                           ctdb->cmd_pnn, TIMEOUT(), db_id,
5657                                           &dbstats);
5658         if (ret != 0) {
5659                 fprintf(stderr, "Failed to get statistics for DB %s\n",
5660                         db_name);
5661                 return ret;
5662         }
5663
5664         print_dbstatistics(db_name, dbstats);
5665         return 0;
5666 }
5667
5668 struct disable_takeover_runs_state {
5669         uint32_t *pnn_list;
5670         int node_count;
5671         bool *reply;
5672         int status;
5673         bool done;
5674 };
5675
5676 static void disable_takeover_run_handler(uint64_t srvid, TDB_DATA data,
5677                                          void *private_data)
5678 {
5679         struct disable_takeover_runs_state *state =
5680                 (struct disable_takeover_runs_state *)private_data;
5681         int ret, i;
5682
5683         if (data.dsize != sizeof(int)) {
5684                 /* Ignore packet */
5685                 return;
5686         }
5687
5688         /* ret will be a PNN (i.e. >=0) on success, or negative on error */
5689         ret = *(int *)data.dptr;
5690         if (ret < 0) {
5691                 state->status = ret;
5692                 state->done = true;
5693                 return;
5694         }
5695         for (i=0; i<state->node_count; i++) {
5696                 if (state->pnn_list[i] == ret) {
5697                         state->reply[i] = true;
5698                         break;
5699                 }
5700         }
5701
5702         state->done = true;
5703         for (i=0; i<state->node_count; i++) {
5704                 if (! state->reply[i]) {
5705                         state->done = false;
5706                         break;
5707                 }
5708         }
5709 }
5710
5711 static int disable_takeover_runs(TALLOC_CTX *mem_ctx,
5712                                  struct ctdb_context *ctdb, uint32_t timeout,
5713                                  uint32_t *pnn_list, int count)
5714 {
5715         struct ctdb_disable_message disable = { 0 };
5716         struct disable_takeover_runs_state state;
5717         int ret, i;
5718
5719         disable.pnn = ctdb->pnn;
5720         disable.srvid = next_srvid(ctdb);
5721         disable.timeout = timeout;
5722
5723         state.pnn_list = pnn_list;
5724         state.node_count = count;
5725         state.done = false;
5726         state.status = 0;
5727         state.reply = talloc_zero_array(mem_ctx, bool, count);
5728         if (state.reply == NULL) {
5729                 return ENOMEM;
5730         }
5731
5732         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
5733                                               disable.srvid,
5734                                               disable_takeover_run_handler,
5735                                               &state);
5736         if (ret != 0) {
5737                 return ret;
5738         }
5739
5740         for (i=0; i<count; i++) {
5741                 ret = ctdb_message_disable_takeover_runs(mem_ctx, ctdb->ev,
5742                                                          ctdb->client,
5743                                                          pnn_list[i],
5744                                                          &disable);
5745                 if (ret != 0) {
5746                         goto fail;
5747                 }
5748         }
5749
5750         ret = ctdb_client_wait_timeout(ctdb->ev, &state.done, TIMEOUT());
5751         if (ret == ETIME) {
5752                 fprintf(stderr, "Timed out waiting to disable takeover runs\n");
5753         } else {
5754                 ret = (state.status >= 0 ? 0 : 1);
5755         }
5756
5757 fail:
5758         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
5759                                            disable.srvid, &state);
5760         return ret;
5761 }
5762
5763 static int control_reloadips(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5764                              int argc, const char **argv)
5765 {
5766         const char *nodestring = NULL;
5767         struct ctdb_node_map *nodemap, *nodemap2;
5768         struct ctdb_req_control request;
5769         uint32_t *pnn_list, *pnn_list2;
5770         int ret, count, count2;
5771
5772         if (argc > 1) {
5773                 usage("reloadips");
5774         }
5775
5776         if (argc == 1) {
5777                 nodestring = argv[0];
5778         }
5779
5780         nodemap = get_nodemap(ctdb, false);
5781         if (nodemap == NULL) {
5782                 return 1;
5783         }
5784
5785         if (! parse_nodestring(mem_ctx, ctdb, nodestring, &nodemap2)) {
5786                 return 1;
5787         }
5788
5789         count = list_of_connected_nodes(nodemap, CTDB_UNKNOWN_PNN,
5790                                         mem_ctx, &pnn_list);
5791         if (count <= 0) {
5792                 fprintf(stderr, "Memory allocation error\n");
5793                 return 1;
5794         }
5795
5796         count2 = list_of_active_nodes(nodemap2, CTDB_UNKNOWN_PNN,
5797                                       mem_ctx, &pnn_list2);
5798         if (count2 <= 0) {
5799                 fprintf(stderr, "Memory allocation error\n");
5800                 return 1;
5801         }
5802
5803         /* Disable takeover runs on all connected nodes.  A reply
5804          * indicating success is needed from each node so all nodes
5805          * will need to be active.
5806          *
5807          * A check could be added to not allow reloading of IPs when
5808          * there are disconnected nodes.  However, this should
5809          * probably be left up to the administrator.
5810          */
5811         ret = disable_takeover_runs(mem_ctx, ctdb, 2*options.timelimit,
5812                                     pnn_list, count);
5813         if (ret != 0) {
5814                 fprintf(stderr, "Failed to disable takeover runs\n");
5815                 return ret;
5816         }
5817
5818         /* Now tell all the desired nodes to reload their public IPs.
5819          * Keep trying this until it succeeds.  This assumes all
5820          * failures are transient, which might not be true...
5821          */
5822         ctdb_req_control_reload_public_ips(&request);
5823         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
5824                                         pnn_list2, count2, TIMEOUT(),
5825                                         &request, NULL, NULL);
5826         if (ret != 0) {
5827                 fprintf(stderr, "Failed to reload IPs on some nodes.\n");
5828         }
5829
5830         /* It isn't strictly necessary to wait until takeover runs are
5831          * re-enabled but doing so can't hurt.
5832          */
5833         ret = disable_takeover_runs(mem_ctx, ctdb, 0, pnn_list, count);
5834         if (ret != 0) {
5835                 fprintf(stderr, "Failed to enable takeover runs\n");
5836                 return ret;
5837         }
5838
5839         return ipreallocate(mem_ctx, ctdb);
5840 }
5841
5842 static int control_ipiface(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5843                            int argc, const char **argv)
5844 {
5845         ctdb_sock_addr addr;
5846         char *iface;
5847
5848         if (argc != 1) {
5849                 usage("ipiface");
5850         }
5851
5852         if (! parse_ip(argv[0], NULL, 0, &addr)) {
5853                 fprintf(stderr, "Failed to Parse IP %s\n", argv[0]);
5854                 return 1;
5855         }
5856
5857         iface = ctdb_sys_find_ifname(&addr);
5858         if (iface == NULL) {
5859                 fprintf(stderr, "Failed to find interface for IP %s\n",
5860                         argv[0]);
5861                 return 1;
5862         }
5863         free(iface);
5864
5865         return 0;
5866 }
5867
5868
5869 static const struct ctdb_cmd {
5870         const char *name;
5871         int (*fn)(TALLOC_CTX *, struct ctdb_context *, int, const char **);
5872         bool without_daemon; /* can be run without daemon running ? */
5873         bool remote; /* can be run on remote nodes */
5874         const char *msg;
5875         const char *args;
5876 } ctdb_commands[] = {
5877         { "version", control_version, true, false,
5878                 "show version of ctdb", NULL },
5879         { "status", control_status, false, true,
5880                 "show node status", NULL },
5881         { "uptime", control_uptime, false, true,
5882                 "show node uptime", NULL },
5883         { "ping", control_ping, false, true,
5884                 "ping all nodes", NULL },
5885         { "runstate", control_runstate, false, true,
5886                 "get/check runstate of a node",
5887                 "[setup|first_recovery|startup|running]" },
5888         { "getvar", control_getvar, false, true,
5889                 "get a tunable variable", "<name>" },
5890         { "setvar", control_setvar, false, true,
5891                 "set a tunable variable", "<name> <value>" },
5892         { "listvars", control_listvars, false, true,
5893                 "list tunable variables", NULL },
5894         { "statistics", control_statistics, false, true,
5895                 "show ctdb statistics", NULL },
5896         { "statisticsreset", control_statistics_reset, false, true,
5897                 "reset ctdb statistics", NULL },
5898         { "stats", control_stats, false, true,
5899                 "show rolling statistics", "[count]" },
5900         { "ip", control_ip, false, true,
5901                 "show public ips", "[all]" },
5902         { "ipinfo", control_ipinfo, false, true,
5903                 "show public ip details", "<ip>" },
5904         { "ifaces", control_ifaces, false, true,
5905                 "show interfaces", NULL },
5906         { "setifacelink", control_setifacelink, false, true,
5907                 "set interface link status", "<iface> up|down" },
5908         { "process-exists", control_process_exists, false, true,
5909                 "check if a process exists on a node",  "<pid> [<srvid>]" },
5910         { "getdbmap", control_getdbmap, false, true,
5911                 "show attached databases", NULL },
5912         { "getdbstatus", control_getdbstatus, false, true,
5913                 "show database status", "<dbname|dbid>" },
5914         { "catdb", control_catdb, false, false,
5915                 "dump cluster-wide ctdb database", "<dbname|dbid>" },
5916         { "cattdb", control_cattdb, false, false,
5917                 "dump local ctdb database", "<dbname|dbid>" },
5918         { "getcapabilities", control_getcapabilities, false, true,
5919                 "show node capabilities", NULL },
5920         { "pnn", control_pnn, false, false,
5921                 "show the pnn of the currnet node", NULL },
5922         { "lvs", control_lvs, false, false,
5923                 "show lvs configuration", "master|list|status" },
5924         { "setdebug", control_setdebug, false, true,
5925                 "set debug level", "ERROR|WARNING|NOTICE|INFO|DEBUG" },
5926         { "getdebug", control_getdebug, false, true,
5927                 "get debug level", NULL },
5928         { "attach", control_attach, false, false,
5929                 "attach a database", "<dbname> [persistent|replicated]" },
5930         { "detach", control_detach, false, false,
5931                 "detach database(s)", "<dbname|dbid> ..." },
5932         { "dumpmemory", control_dumpmemory, false, true,
5933                 "dump ctdbd memory map", NULL },
5934         { "rddumpmemory", control_rddumpmemory, false, true,
5935                 "dump recoverd memory map", NULL },
5936         { "getpid", control_getpid, false, true,
5937                 "get ctdbd process ID", NULL },
5938         { "disable", control_disable, false, true,
5939                 "disable a node", NULL },
5940         { "enable", control_enable, false, true,
5941                 "enable a node", NULL },
5942         { "stop", control_stop, false, true,
5943                 "stop a node", NULL },
5944         { "continue", control_continue, false, true,
5945                 "continue a stopped node", NULL },
5946         { "ban", control_ban, false, true,
5947                 "ban a node", "<bantime>"},
5948         { "unban", control_unban, false, true,
5949                 "unban a node", NULL },
5950         { "shutdown", control_shutdown, false, true,
5951                 "shutdown ctdb daemon", NULL },
5952         { "recover", control_recover, false, true,
5953                 "force recovery", NULL },
5954         { "sync", control_ipreallocate, false, true,
5955                 "run ip reallocation (deprecated)", NULL },
5956         { "ipreallocate", control_ipreallocate, false, true,
5957                 "run ip reallocation", NULL },
5958         { "isnotrecmaster", control_isnotrecmaster, false, false,
5959                 "check if local node is the recmaster", NULL },
5960         { "gratarp", control_gratarp, false, true,
5961                 "send a gratuitous arp", "<ip> <interface>" },
5962         { "tickle", control_tickle, true, false,
5963                 "send a tcp tickle ack", "<srcip:port> <dstip:port>" },
5964         { "gettickles", control_gettickles, false, true,
5965                 "get the list of tickles", "<ip> [<port>]" },
5966         { "addtickle", control_addtickle, false, true,
5967                 "add a tickle", "<ip>:<port> <ip>:<port>" },
5968         { "deltickle", control_deltickle, false, true,
5969                 "delete a tickle", "<ip>:<port> <ip>:<port>" },
5970         { "listnodes", control_listnodes, true, true,
5971                 "list nodes in the cluster", NULL },
5972         { "reloadnodes", control_reloadnodes, false, false,
5973                 "reload the nodes file all nodes", NULL },
5974         { "moveip", control_moveip, false, false,
5975                 "move an ip address to another node", "<ip> <node>" },
5976         { "addip", control_addip, false, true,
5977                 "add an ip address to a node", "<ip/mask> <iface>" },
5978         { "delip", control_delip, false, true,
5979                 "delete an ip address from a node", "<ip>" },
5980         { "backupdb", control_backupdb, false, false,
5981                 "backup a database into a file", "<dbname|dbid> <file>" },
5982         { "restoredb", control_restoredb, false, false,
5983                 "restore a database from a file", "<file> [dbname]" },
5984         { "dumpdbbackup", control_dumpdbbackup, true, false,
5985                 "dump database from a backup file", "<file>" },
5986         { "wipedb", control_wipedb, false, false,
5987                 "wipe the contents of a database.", "<dbname|dbid>"},
5988         { "recmaster", control_recmaster, false, true,
5989                 "show the pnn for the recovery master", NULL },
5990         { "event", control_event, true, false,
5991                 "event and event script commands", NULL },
5992         { "scriptstatus", control_scriptstatus, true, false,
5993                 "show event script status",
5994                 "[init|setup|startup|monitor|takeip|releaseip|ipreallocated]" },
5995         { "natgw", control_natgw, false, false,
5996                 "show natgw configuration", "master|list|status" },
5997         { "getreclock", control_getreclock, false, true,
5998                 "get recovery lock file", NULL },
5999         { "setlmasterrole", control_setlmasterrole, false, true,
6000                 "set LMASTER role", "on|off" },
6001         { "setrecmasterrole", control_setrecmasterrole, false, true,
6002                 "set RECMASTER role", "on|off"},
6003         { "setdbreadonly", control_setdbreadonly, false, true,
6004                 "enable readonly records", "<dbname|dbid>" },
6005         { "setdbsticky", control_setdbsticky, false, true,
6006                 "enable sticky records", "<dbname|dbid>"},
6007         { "pfetch", control_pfetch, false, false,
6008                 "fetch record from persistent database", "<dbname|dbid> <key> [<file>]" },
6009         { "pstore", control_pstore, false, false,
6010                 "write record to persistent database", "<dbname|dbid> <key> <value>" },
6011         { "pdelete", control_pdelete, false, false,
6012                 "delete record from persistent database", "<dbname|dbid> <key>" },
6013         { "ptrans", control_ptrans, false, false,
6014                 "update a persistent database (from file or stdin)", "<dbname|dbid> [<file>]" },
6015         { "tfetch", control_tfetch, false, true,
6016                 "fetch a record", "<tdb-file> <key> [<file>]" },
6017         { "tstore", control_tstore, false, true,
6018                 "store a record", "<tdb-file> <key> <data> [<rsn> <dmaster> <flags>]" },
6019         { "readkey", control_readkey, false, false,
6020                 "read value of a database key", "<dbname|dbid> <key> [readonly]" },
6021         { "writekey", control_writekey, false, false,
6022                 "write value for a database key", "<dbname|dbid> <key> <value>" },
6023         { "deletekey", control_deletekey, false, false,
6024                 "delete a database key", "<dbname|dbid> <key>" },
6025         { "checktcpport", control_checktcpport, true, false,
6026                 "check if a service is bound to a specific tcp port or not", "<port>" },
6027         { "getdbseqnum", control_getdbseqnum, false, false,
6028                 "get database sequence number", "<dbname|dbid>" },
6029         { "nodestatus", control_nodestatus, false, true,
6030                 "show and return node status", "[all|<pnn-list>]" },
6031         { "dbstatistics", control_dbstatistics, false, true,
6032                 "show database statistics", "<dbname|dbid>" },
6033         { "reloadips", control_reloadips, false, false,
6034                 "reload the public addresses file", "[all|<pnn-list>]" },
6035         { "ipiface", control_ipiface, true, false,
6036                 "Find the interface an ip address is hosted on", "<ip>" },
6037 };
6038
6039 static const struct ctdb_cmd *match_command(const char *command)
6040 {
6041         const struct ctdb_cmd *cmd;
6042         int i;
6043
6044         for (i=0; i<ARRAY_SIZE(ctdb_commands); i++) {
6045                 cmd = &ctdb_commands[i];
6046                 if (strlen(command) == strlen(cmd->name) &&
6047                     strncmp(command, cmd->name, strlen(command)) == 0) {
6048                         return cmd;
6049                 }
6050         }
6051
6052         return NULL;
6053 }
6054
6055
6056 /**
6057  * Show usage message
6058  */
6059 static void usage_full(void)
6060 {
6061         int i;
6062
6063         poptPrintHelp(pc, stdout, 0);
6064         printf("\nCommands:\n");
6065         for (i=0; i<ARRAY_SIZE(ctdb_commands); i++) {
6066                 printf("  %-15s %-27s  %s\n",
6067                        ctdb_commands[i].name,
6068                        ctdb_commands[i].args ? ctdb_commands[i].args : "",
6069                        ctdb_commands[i].msg);
6070         }
6071 }
6072
6073 static void usage(const char *command)
6074 {
6075         const struct ctdb_cmd *cmd;
6076
6077         if (command == NULL) {
6078                 usage_full();
6079                 exit(1);
6080         }
6081
6082         cmd = match_command(command);
6083         if (cmd == NULL) {
6084                 usage_full();
6085         } else {
6086                 poptPrintUsage(pc, stdout, 0);
6087                 printf("\nCommands:\n");
6088                 printf("  %-15s %-27s  %s\n",
6089                        cmd->name, cmd->args ? cmd->args : "", cmd->msg);
6090         }
6091
6092         exit(1);
6093 }
6094
6095 struct poptOption cmdline_options[] = {
6096         POPT_AUTOHELP
6097         { "socket", 's', POPT_ARG_STRING, &options.socket, 0,
6098                 "CTDB socket path", "filename" },
6099         { "debug", 'd', POPT_ARG_STRING, &options.debuglevelstr, 0,
6100                 "debug level"},
6101         { "timelimit", 't', POPT_ARG_INT, &options.timelimit, 0,
6102                 "timelimit (in seconds)" },
6103         { "node", 'n', POPT_ARG_INT, &options.pnn, 0,
6104                 "node specification - integer" },
6105         { NULL, 'Y', POPT_ARG_NONE, &options.machinereadable, 0,
6106                 "enable machine readable output", NULL },
6107         { "separator", 'x', POPT_ARG_STRING, &options.sep, 0,
6108                 "specify separator for machine readable output", "CHAR" },
6109         { NULL, 'X', POPT_ARG_NONE, &options.machineparsable, 0,
6110                 "enable machine parsable output with separator |", NULL },
6111         { "verbose", 'v', POPT_ARG_NONE, &options.verbose, 0,
6112                 "enable verbose output", NULL },
6113         { "maxruntime", 'T', POPT_ARG_INT, &options.maxruntime, 0,
6114                 "die if runtime exceeds this limit (in seconds)" },
6115         POPT_TABLEEND
6116 };
6117
6118 static int process_command(const struct ctdb_cmd *cmd, int argc,
6119                            const char **argv)
6120 {
6121         TALLOC_CTX *tmp_ctx;
6122         struct ctdb_context *ctdb;
6123         int ret;
6124         bool status;
6125         uint64_t srvid_offset;
6126
6127         tmp_ctx = talloc_new(NULL);
6128         if (tmp_ctx == NULL) {
6129                 fprintf(stderr, "Memory allocation error\n");
6130                 goto fail;
6131         }
6132
6133         if (cmd->without_daemon) {
6134                 if (options.pnn != -1) {
6135                         fprintf(stderr,
6136                                 "Cannot specify node for command %s\n",
6137                                 cmd->name);
6138                         goto fail;
6139                 }
6140
6141                 ret = cmd->fn(tmp_ctx, NULL, argc-1, argv+1);
6142                 talloc_free(tmp_ctx);
6143                 return ret;
6144         }
6145
6146         ctdb = talloc_zero(tmp_ctx, struct ctdb_context);
6147         if (ctdb == NULL) {
6148                 fprintf(stderr, "Memory allocation error\n");
6149                 goto fail;
6150         }
6151
6152         ctdb->ev = tevent_context_init(ctdb);
6153         if (ctdb->ev == NULL) {
6154                 fprintf(stderr, "Failed to initialize tevent\n");
6155                 goto fail;
6156         }
6157
6158         ret = ctdb_client_init(ctdb, ctdb->ev, options.socket, &ctdb->client);
6159         if (ret != 0) {
6160                 fprintf(stderr, "Failed to connect to CTDB daemon (%s)\n",
6161                         options.socket);
6162
6163                 if (!find_node_xpnn(ctdb, NULL)) {
6164                         fprintf(stderr, "Is this node part of CTDB cluster?\n");
6165                 }
6166                 goto fail;
6167         }
6168
6169         ctdb->pnn = ctdb_client_pnn(ctdb->client);
6170         srvid_offset = getpid() & 0xFFFF;
6171         ctdb->srvid = SRVID_CTDB_TOOL | (srvid_offset << 16);
6172
6173         if (options.pnn != -1) {
6174                 status = verify_pnn(ctdb, options.pnn);
6175                 if (! status) {
6176                         goto fail;
6177                 }
6178
6179                 ctdb->cmd_pnn = options.pnn;
6180         } else {
6181                 ctdb->cmd_pnn = ctdb->pnn;
6182         }
6183
6184         if (! cmd->remote && ctdb->pnn != ctdb->cmd_pnn) {
6185                 fprintf(stderr, "Node cannot be specified for command %s\n",
6186                         cmd->name);
6187                 goto fail;
6188         }
6189
6190         ret = cmd->fn(tmp_ctx, ctdb, argc-1, argv+1);
6191         talloc_free(tmp_ctx);
6192         return ret;
6193
6194 fail:
6195         talloc_free(tmp_ctx);
6196         return 1;
6197 }
6198
6199 static void signal_handler(int sig)
6200 {
6201         fprintf(stderr, "Maximum runtime exceeded - exiting\n");
6202 }
6203
6204 static void alarm_handler(int sig)
6205 {
6206         /* Kill any child processes */
6207         signal(SIGTERM, signal_handler);
6208         kill(0, SIGTERM);
6209
6210         _exit(1);
6211 }
6212
6213 int main(int argc, const char *argv[])
6214 {
6215         int opt;
6216         const char **extra_argv;
6217         int extra_argc;
6218         const struct ctdb_cmd *cmd;
6219         const char *ctdb_socket;
6220         int loglevel;
6221         int ret;
6222
6223         setlinebuf(stdout);
6224
6225         /* Set default options */
6226         options.socket = CTDB_SOCKET;
6227         options.debuglevelstr = NULL;
6228         options.timelimit = 10;
6229         options.sep = "|";
6230         options.maxruntime = 0;
6231         options.pnn = -1;
6232
6233         ctdb_socket = getenv("CTDB_SOCKET");
6234         if (ctdb_socket != NULL) {
6235                 options.socket = ctdb_socket;
6236         }
6237
6238         pc = poptGetContext(argv[0], argc, argv, cmdline_options,
6239                             POPT_CONTEXT_KEEP_FIRST);
6240         while ((opt = poptGetNextOpt(pc)) != -1) {
6241                 fprintf(stderr, "Invalid option %s: %s\n",
6242                         poptBadOption(pc, 0), poptStrerror(opt));
6243                 exit(1);
6244         }
6245
6246         if (options.maxruntime == 0) {
6247                 const char *ctdb_timeout;
6248
6249                 ctdb_timeout = getenv("CTDB_TIMEOUT");
6250                 if (ctdb_timeout != NULL) {
6251                         options.maxruntime = strtoul(ctdb_timeout, NULL, 0);
6252                 } else {
6253                         options.maxruntime = 120;
6254                 }
6255         }
6256         if (options.maxruntime <= 120) {
6257                 /* default timeout is 120 seconds */
6258                 options.maxruntime = 120;
6259         }
6260
6261         if (options.machineparsable) {
6262                 options.machinereadable = 1;
6263         }
6264
6265         /* setup the remaining options for the commands */
6266         extra_argc = 0;
6267         extra_argv = poptGetArgs(pc);
6268         if (extra_argv) {
6269                 extra_argv++;
6270                 while (extra_argv[extra_argc]) extra_argc++;
6271         }
6272
6273         if (extra_argc < 1) {
6274                 usage(NULL);
6275         }
6276
6277         cmd = match_command(extra_argv[0]);
6278         if (cmd == NULL) {
6279                 fprintf(stderr, "Unknown command '%s'\n", extra_argv[0]);
6280                 exit(1);
6281         }
6282
6283         /* Enable logging */
6284         setup_logging("ctdb", DEBUG_STDERR);
6285         if (debug_level_parse(options.debuglevelstr, &loglevel)) {
6286                 DEBUGLEVEL = loglevel;
6287         } else {
6288                 DEBUGLEVEL = DEBUG_ERR;
6289         }
6290
6291         signal(SIGALRM, alarm_handler);
6292         alarm(options.maxruntime);
6293
6294         ret = process_command(cmd, extra_argc, extra_argv);
6295         if (ret == -1) {
6296                 ret = 1;
6297         }
6298
6299         (void)poptFreeContext(pc);
6300
6301         return ret;
6302 }