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         int ret, status;
1869
1870         if (argc != 1) {
1871                 usage("process-exists");
1872         }
1873
1874         pid = atoi(argv[0]);
1875         ret = ctdb_ctrl_process_exists(mem_ctx, ctdb->ev, ctdb->client,
1876                                        ctdb->cmd_pnn, TIMEOUT(), pid, &status);
1877         if (ret != 0) {
1878                 return ret;
1879         }
1880
1881         if (status == 0) {
1882                 printf("PID %u exists\n", pid);
1883         } else {
1884                 printf("PID %u does not exist\n", pid);
1885         }
1886         return status;
1887 }
1888
1889 static int control_getdbmap(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1890                             int argc, const char **argv)
1891 {
1892         struct ctdb_dbid_map *dbmap;
1893         int ret, i;
1894
1895         if (argc != 0) {
1896                 usage("getdbmap");
1897         }
1898
1899         ret = ctdb_ctrl_get_dbmap(mem_ctx, ctdb->ev, ctdb->client,
1900                                   ctdb->cmd_pnn, TIMEOUT(), &dbmap);
1901         if (ret != 0) {
1902                 return ret;
1903         }
1904
1905         if (options.machinereadable == 1) {
1906                 printf("%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s\n",
1907                        options.sep,
1908                        "ID", options.sep,
1909                        "Name", options.sep,
1910                        "Path", options.sep,
1911                        "Persistent", options.sep,
1912                        "Sticky", options.sep,
1913                        "Unhealthy", options.sep,
1914                        "Readonly", options.sep,
1915                        "Replicated", options.sep);
1916         } else {
1917                 printf("Number of databases:%d\n", dbmap->num);
1918         }
1919
1920         for (i=0; i<dbmap->num; i++) {
1921                 const char *name;
1922                 const char *path;
1923                 const char *health;
1924                 bool persistent;
1925                 bool readonly;
1926                 bool sticky;
1927                 bool replicated;
1928                 uint32_t db_id;
1929
1930                 db_id = dbmap->dbs[i].db_id;
1931
1932                 ret = ctdb_ctrl_get_dbname(mem_ctx, ctdb->ev, ctdb->client,
1933                                            ctdb->cmd_pnn, TIMEOUT(), db_id,
1934                                            &name);
1935                 if (ret != 0) {
1936                         return ret;
1937                 }
1938
1939                 ret = ctdb_ctrl_getdbpath(mem_ctx, ctdb->ev, ctdb->client,
1940                                           ctdb->cmd_pnn, TIMEOUT(), db_id,
1941                                           &path);
1942                 if (ret != 0) {
1943                         return ret;
1944                 }
1945
1946                 ret = ctdb_ctrl_db_get_health(mem_ctx, ctdb->ev, ctdb->client,
1947                                               ctdb->cmd_pnn, TIMEOUT(), db_id,
1948                                               &health);
1949                 if (ret != 0) {
1950                         return ret;
1951                 }
1952
1953                 persistent = dbmap->dbs[i].flags & CTDB_DB_FLAGS_PERSISTENT;
1954                 readonly = dbmap->dbs[i].flags & CTDB_DB_FLAGS_READONLY;
1955                 sticky = dbmap->dbs[i].flags & CTDB_DB_FLAGS_STICKY;
1956                 replicated = dbmap->dbs[i].flags & CTDB_DB_FLAGS_REPLICATED;
1957
1958                 if (options.machinereadable == 1) {
1959                         printf("%s0x%08X%s%s%s%s%s%d%s%d%s%d%s%d%s%d%s\n",
1960                                options.sep,
1961                                db_id, options.sep,
1962                                name, options.sep,
1963                                path, options.sep,
1964                                !! (persistent), options.sep,
1965                                !! (sticky), options.sep,
1966                                !! (health), options.sep,
1967                                !! (readonly), options.sep,
1968                                !! (replicated), options.sep);
1969                 } else {
1970                         printf("dbid:0x%08x name:%s path:%s%s%s%s%s%s\n",
1971                                db_id, name, path,
1972                                persistent ? " PERSISTENT" : "",
1973                                sticky ? " STICKY" : "",
1974                                readonly ? " READONLY" : "",
1975                                replicated ? " REPLICATED" : "",
1976                                health ? " UNHEALTHY" : "");
1977                 }
1978
1979                 talloc_free(discard_const(name));
1980                 talloc_free(discard_const(path));
1981                 talloc_free(discard_const(health));
1982         }
1983
1984         return 0;
1985 }
1986
1987 static int control_getdbstatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
1988                                int argc, const char **argv)
1989 {
1990         uint32_t db_id;
1991         const char *db_name, *db_path, *db_health;
1992         uint8_t db_flags;
1993         int ret;
1994
1995         if (argc != 1) {
1996                 usage("getdbstatus");
1997         }
1998
1999         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
2000                 return 1;
2001         }
2002
2003         ret = ctdb_ctrl_getdbpath(mem_ctx, ctdb->ev, ctdb->client,
2004                                   ctdb->cmd_pnn, TIMEOUT(), db_id,
2005                                   &db_path);
2006         if (ret != 0) {
2007                 return ret;
2008         }
2009
2010         ret = ctdb_ctrl_db_get_health(mem_ctx, ctdb->ev, ctdb->client,
2011                                       ctdb->cmd_pnn, TIMEOUT(), db_id,
2012                                       &db_health);
2013         if (ret != 0) {
2014                 return ret;
2015         }
2016
2017         printf("dbid: 0x%08x\nname: %s\npath: %s\n", db_id, db_name, db_path);
2018         printf("PERSISTENT: %s\nREPLICATED: %s\nSTICKY: %s\nREADONLY: %s\n",
2019                (db_flags & CTDB_DB_FLAGS_PERSISTENT ? "yes" : "no"),
2020                (db_flags & CTDB_DB_FLAGS_REPLICATED ? "yes" : "no"),
2021                (db_flags & CTDB_DB_FLAGS_STICKY ? "yes" : "no"),
2022                (db_flags & CTDB_DB_FLAGS_READONLY ? "yes" : "no"));
2023         printf("HEALTH: %s\n", (db_health ? db_health : "OK"));
2024         return 0;
2025 }
2026
2027 struct dump_record_state {
2028         uint32_t count;
2029 };
2030
2031 #define ISASCII(x) (isprint(x) && ! strchr("\"\\", (x)))
2032
2033 static void dump_tdb_data(const char *name, TDB_DATA val)
2034 {
2035         int i;
2036
2037         fprintf(stdout, "%s(%zu) = \"", name, val.dsize);
2038         for (i=0; i<val.dsize; i++) {
2039                 if (ISASCII(val.dptr[i])) {
2040                         fprintf(stdout, "%c", val.dptr[i]);
2041                 } else {
2042                         fprintf(stdout, "\\%02X", val.dptr[i]);
2043                 }
2044         }
2045         fprintf(stdout, "\"\n");
2046 }
2047
2048 static void dump_ltdb_header(struct ctdb_ltdb_header *header)
2049 {
2050         fprintf(stdout, "dmaster: %u\n", header->dmaster);
2051         fprintf(stdout, "rsn: %" PRIu64 "\n", header->rsn);
2052         fprintf(stdout, "flags: 0x%08x", header->flags);
2053         if (header->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) {
2054                 fprintf(stdout, " MIGRATED_WITH_DATA");
2055         }
2056         if (header->flags & CTDB_REC_FLAG_VACUUM_MIGRATED) {
2057                 fprintf(stdout, " VACUUM_MIGRATED");
2058         }
2059         if (header->flags & CTDB_REC_FLAG_AUTOMATIC) {
2060                 fprintf(stdout, " AUTOMATIC");
2061         }
2062         if (header->flags & CTDB_REC_RO_HAVE_DELEGATIONS) {
2063                 fprintf(stdout, " RO_HAVE_DELEGATIONS");
2064         }
2065         if (header->flags & CTDB_REC_RO_HAVE_READONLY) {
2066                 fprintf(stdout, " RO_HAVE_READONLY");
2067         }
2068         if (header->flags & CTDB_REC_RO_REVOKING_READONLY) {
2069                 fprintf(stdout, " RO_REVOKING_READONLY");
2070         }
2071         if (header->flags & CTDB_REC_RO_REVOKE_COMPLETE) {
2072                 fprintf(stdout, " RO_REVOKE_COMPLETE");
2073         }
2074         fprintf(stdout, "\n");
2075
2076 }
2077
2078 static int dump_record(uint32_t reqid, struct ctdb_ltdb_header *header,
2079                        TDB_DATA key, TDB_DATA data, void *private_data)
2080 {
2081         struct dump_record_state *state =
2082                 (struct dump_record_state *)private_data;
2083
2084         state->count += 1;
2085
2086         dump_tdb_data("key", key);
2087         dump_ltdb_header(header);
2088         dump_tdb_data("data", data);
2089         fprintf(stdout, "\n");
2090
2091         return 0;
2092 }
2093
2094 static int control_catdb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2095                          int argc, const char **argv)
2096 {
2097         struct ctdb_db_context *db;
2098         const char *db_name;
2099         uint32_t db_id;
2100         uint8_t db_flags;
2101         struct dump_record_state state;
2102         int ret;
2103
2104         if (argc != 1) {
2105                 usage("catdb");
2106         }
2107
2108         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
2109                 return 1;
2110         }
2111
2112         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
2113                           db_flags, &db);
2114         if (ret != 0) {
2115                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
2116                 return ret;
2117         }
2118
2119         state.count = 0;
2120
2121         ret = ctdb_db_traverse(mem_ctx, ctdb->ev, ctdb->client, db,
2122                                ctdb->cmd_pnn, TIMEOUT(),
2123                                dump_record, &state);
2124
2125         printf("Dumped %u records\n", state.count);
2126
2127         return ret;
2128 }
2129
2130 static int control_cattdb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2131                           int argc, const char **argv)
2132 {
2133         struct ctdb_db_context *db;
2134         const char *db_name;
2135         uint32_t db_id;
2136         uint8_t db_flags;
2137         struct dump_record_state state;
2138         int ret;
2139
2140         if (argc != 1) {
2141                 usage("catdb");
2142         }
2143
2144         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
2145                 return 1;
2146         }
2147
2148         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
2149                           db_flags, &db);
2150         if (ret != 0) {
2151                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
2152                 return ret;
2153         }
2154
2155         state.count = 0;
2156         ret = ctdb_db_traverse_local(db, true, true, dump_record, &state);
2157
2158         printf("Dumped %u record(s)\n", state.count);
2159
2160         return ret;
2161 }
2162
2163 static int control_getcapabilities(TALLOC_CTX *mem_ctx,
2164                                    struct ctdb_context *ctdb,
2165                                    int argc, const char **argv)
2166 {
2167         uint32_t caps;
2168         int ret;
2169
2170         if (argc != 0) {
2171                 usage("getcapabilities");
2172         }
2173
2174         ret = ctdb_ctrl_get_capabilities(mem_ctx, ctdb->ev, ctdb->client,
2175                                          ctdb->cmd_pnn, TIMEOUT(), &caps);
2176         if (ret != 0) {
2177                 return ret;
2178         }
2179
2180         if (options.machinereadable == 1) {
2181                 printf("%s%s%s%s%s\n",
2182                        options.sep,
2183                        "RECMASTER", options.sep,
2184                        "LMASTER", options.sep);
2185                 printf("%s%d%s%d%s\n", options.sep,
2186                        !! (caps & CTDB_CAP_RECMASTER), options.sep,
2187                        !! (caps & CTDB_CAP_LMASTER), options.sep);
2188         } else {
2189                 printf("RECMASTER: %s\n",
2190                        (caps & CTDB_CAP_RECMASTER) ? "YES" : "NO");
2191                 printf("LMASTER: %s\n",
2192                        (caps & CTDB_CAP_LMASTER) ? "YES" : "NO");
2193         }
2194
2195         return 0;
2196 }
2197
2198 static int control_pnn(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2199                        int argc, const char **argv)
2200 {
2201         printf("%u\n", ctdb_client_pnn(ctdb->client));
2202         return 0;
2203 }
2204
2205 static int control_lvs(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2206                        int argc, const char **argv)
2207 {
2208         char *t, *lvs_helper = NULL;
2209
2210         if (argc != 1) {
2211                 usage("lvs");
2212         }
2213
2214         t = getenv("CTDB_LVS_HELPER");
2215         if (t != NULL) {
2216                 lvs_helper = talloc_strdup(mem_ctx, t);
2217         } else {
2218                 lvs_helper = talloc_asprintf(mem_ctx, "%s/ctdb_lvs",
2219                                              CTDB_HELPER_BINDIR);
2220         }
2221
2222         if (lvs_helper == NULL) {
2223                 fprintf(stderr, "Unable to set LVS helper\n");
2224                 return 1;
2225         }
2226
2227         return run_helper(mem_ctx, "LVS helper", lvs_helper, argc, argv);
2228 }
2229
2230 static int control_setdebug(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2231                             int argc, const char **argv)
2232 {
2233         int log_level;
2234         int ret;
2235         bool found;
2236
2237         if (argc != 1) {
2238                 usage("setdebug");
2239         }
2240
2241         found = debug_level_parse(argv[0], &log_level);
2242         if (! found) {
2243                 fprintf(stderr,
2244                         "Invalid debug level '%s'. Valid levels are:\n",
2245                         argv[0]);
2246                 fprintf(stderr, "\tERROR | WARNING | NOTICE | INFO | DEBUG\n");
2247                 return 1;
2248         }
2249
2250         ret = ctdb_ctrl_setdebug(mem_ctx, ctdb->ev, ctdb->client,
2251                                  ctdb->cmd_pnn, TIMEOUT(), log_level);
2252         if (ret != 0) {
2253                 return ret;
2254         }
2255
2256         return 0;
2257 }
2258
2259 static int control_getdebug(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2260                             int argc, const char **argv)
2261 {
2262         int loglevel;
2263         const char *log_str;
2264         int ret;
2265
2266         if (argc != 0) {
2267                 usage("getdebug");
2268         }
2269
2270         ret = ctdb_ctrl_getdebug(mem_ctx, ctdb->ev, ctdb->client,
2271                                  ctdb->cmd_pnn, TIMEOUT(), &loglevel);
2272         if (ret != 0) {
2273                 return ret;
2274         }
2275
2276         log_str = debug_level_to_string(loglevel);
2277         printf("%s\n", log_str);
2278
2279         return 0;
2280 }
2281
2282 static int control_attach(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2283                           int argc, const char **argv)
2284 {
2285         const char *db_name;
2286         uint8_t db_flags = 0;
2287         int ret;
2288
2289         if (argc < 1 || argc > 2) {
2290                 usage("attach");
2291         }
2292
2293         db_name = argv[0];
2294         if (argc == 2) {
2295                 if (strcmp(argv[1], "persistent") == 0) {
2296                         db_flags = CTDB_DB_FLAGS_PERSISTENT;
2297                 } else if (strcmp(argv[1], "readonly") == 0) {
2298                         db_flags = CTDB_DB_FLAGS_READONLY;
2299                 } else if (strcmp(argv[1], "sticky") == 0) {
2300                         db_flags = CTDB_DB_FLAGS_STICKY;
2301                 } else if (strcmp(argv[1], "replicated") == 0) {
2302                         db_flags = CTDB_DB_FLAGS_REPLICATED;
2303                 } else {
2304                         usage("attach");
2305                 }
2306         }
2307
2308         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
2309                           db_flags, NULL);
2310         if (ret != 0) {
2311                 return ret;
2312         }
2313
2314         return 0;
2315 }
2316
2317 static int control_detach(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2318                           int argc, const char **argv)
2319 {
2320         const char *db_name;
2321         uint32_t db_id;
2322         uint8_t db_flags;
2323         struct ctdb_node_map *nodemap;
2324         int recmode;
2325         int ret, ret2, i;
2326
2327         if (argc < 1) {
2328                 usage("detach");
2329         }
2330
2331         ret = ctdb_ctrl_get_recmode(mem_ctx, ctdb->ev, ctdb->client,
2332                                     ctdb->cmd_pnn, TIMEOUT(), &recmode);
2333         if (ret != 0) {
2334                 return ret;
2335         }
2336
2337         if (recmode == CTDB_RECOVERY_ACTIVE) {
2338                 fprintf(stderr, "Database cannot be detached"
2339                                 " when recovery is active\n");
2340                 return 1;
2341         }
2342
2343         nodemap = get_nodemap(ctdb, false);
2344         if (nodemap == NULL) {
2345                 return 1;
2346         }
2347
2348         for (i=0; i<nodemap->num; i++) {
2349                 uint32_t value;
2350
2351                 if (nodemap->node[i].flags & NODE_FLAGS_DISCONNECTED) {
2352                         continue;
2353                 }
2354                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
2355                         continue;
2356                 }
2357                 if (nodemap->node[i].flags & NODE_FLAGS_INACTIVE) {
2358                         fprintf(stderr, "Database cannot be detached on"
2359                                 " inactive (stopped or banned) node %u\n",
2360                                 nodemap->node[i].pnn);
2361                         return 1;
2362                 }
2363
2364                 ret = ctdb_ctrl_get_tunable(mem_ctx, ctdb->ev, ctdb->client,
2365                                             nodemap->node[i].pnn, TIMEOUT(),
2366                                             "AllowClientDBAttach", &value);
2367                 if (ret != 0) {
2368                         fprintf(stderr,
2369                                 "Unable to get tunable AllowClientDBAttach"
2370                                 " from node %u\n", nodemap->node[i].pnn);
2371                         return ret;
2372                 }
2373
2374                 if (value == 1) {
2375                         fprintf(stderr,
2376                                 "Database access is still active on node %u."
2377                                 " Set AllowclientDBAttach=0 on all nodes.\n",
2378                                 nodemap->node[i].pnn);
2379                         return 1;
2380                 }
2381         }
2382
2383         ret2 = 0;
2384         for (i=0; i<argc; i++) {
2385                 if (! db_exists(mem_ctx, ctdb, argv[i], &db_id, &db_name,
2386                                 &db_flags)) {
2387                         continue;
2388                 }
2389
2390                 if (db_flags &
2391                     (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
2392                         fprintf(stderr,
2393                                 "Only volatile databases can be detached\n");
2394                         return 1;
2395                 }
2396
2397                 ret = ctdb_detach(ctdb->ev, ctdb->client, TIMEOUT(), db_id);
2398                 if (ret != 0) {
2399                         fprintf(stderr, "Database %s detach failed\n", db_name);
2400                         ret2 = ret;
2401                 }
2402         }
2403
2404         return ret2;
2405 }
2406
2407 static int control_dumpmemory(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2408                               int argc, const char **argv)
2409 {
2410         const char *mem_str;
2411         ssize_t n;
2412         int ret;
2413
2414         ret = ctdb_ctrl_dump_memory(mem_ctx, ctdb->ev, ctdb->client,
2415                                     ctdb->cmd_pnn, TIMEOUT(), &mem_str);
2416         if (ret != 0) {
2417                 return ret;
2418         }
2419
2420         n = write(1, mem_str, strlen(mem_str)+1);
2421         if (n < 0 || n != strlen(mem_str)+1) {
2422                 fprintf(stderr, "Failed to write talloc summary\n");
2423                 return 1;
2424         }
2425
2426         return 0;
2427 }
2428
2429 static void dump_memory(uint64_t srvid, TDB_DATA data, void *private_data)
2430 {
2431         bool *done = (bool *)private_data;
2432         ssize_t n;
2433
2434         n = write(1, data.dptr, data.dsize);
2435         if (n < 0 || n != data.dsize) {
2436                 fprintf(stderr, "Failed to write talloc summary\n");
2437         }
2438
2439         *done = true;
2440 }
2441
2442 static int control_rddumpmemory(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2443                                 int argc, const char **argv)
2444 {
2445         struct ctdb_srvid_message msg = { 0 };
2446         int ret;
2447         bool done = false;
2448
2449         msg.pnn = ctdb->pnn;
2450         msg.srvid = next_srvid(ctdb);
2451
2452         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
2453                                               msg.srvid, dump_memory, &done);
2454         if (ret != 0) {
2455                 return ret;
2456         }
2457
2458         ret = ctdb_message_mem_dump(mem_ctx, ctdb->ev, ctdb->client,
2459                                     ctdb->cmd_pnn, &msg);
2460         if (ret != 0) {
2461                 return ret;
2462         }
2463
2464         ctdb_client_wait(ctdb->ev, &done);
2465         return 0;
2466 }
2467
2468 static int control_getpid(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2469                           int argc, const char **argv)
2470 {
2471         pid_t pid;
2472         int ret;
2473
2474         ret = ctdb_ctrl_get_pid(mem_ctx, ctdb->ev, ctdb->client,
2475                                 ctdb->cmd_pnn, TIMEOUT(), &pid);
2476         if (ret != 0) {
2477                 return ret;
2478         }
2479
2480         printf("%u\n", pid);
2481         return 0;
2482 }
2483
2484 static int check_flags(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2485                        const char *desc, uint32_t flag, bool set_flag)
2486 {
2487         struct ctdb_node_map *nodemap;
2488         bool flag_is_set;
2489
2490         nodemap = get_nodemap(ctdb, false);
2491         if (nodemap == NULL) {
2492                 return 1;
2493         }
2494
2495         flag_is_set = nodemap->node[ctdb->cmd_pnn].flags & flag;
2496         if (set_flag == flag_is_set) {
2497                 if (set_flag) {
2498                         fprintf(stderr, "Node %u is already %s\n",
2499                                 ctdb->cmd_pnn, desc);
2500                 } else {
2501                         fprintf(stderr, "Node %u is not %s\n",
2502                                 ctdb->cmd_pnn, desc);
2503                 }
2504                 return 0;
2505         }
2506
2507         return 1;
2508 }
2509
2510 static void wait_for_flags(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2511                            uint32_t flag, bool set_flag)
2512 {
2513         struct ctdb_node_map *nodemap;
2514         bool flag_is_set;
2515
2516         while (1) {
2517                 nodemap = get_nodemap(ctdb, true);
2518                 if (nodemap == NULL) {
2519                         fprintf(stderr,
2520                                 "Failed to get nodemap, trying again\n");
2521                         sleep(1);
2522                         continue;
2523                 }
2524
2525                 flag_is_set = nodemap->node[ctdb->cmd_pnn].flags & flag;
2526                 if (flag_is_set == set_flag) {
2527                         break;
2528                 }
2529
2530                 sleep(1);
2531         }
2532 }
2533
2534 static int ctdb_ctrl_modflags(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
2535                               struct ctdb_client_context *client,
2536                               uint32_t destnode, struct timeval timeout,
2537                               uint32_t set, uint32_t clear)
2538 {
2539         struct ctdb_node_map *nodemap;
2540         struct ctdb_node_flag_change flag_change;
2541         struct ctdb_req_control request;
2542         uint32_t *pnn_list;
2543         int ret, count;
2544
2545         ret = ctdb_ctrl_get_nodemap(mem_ctx, ev, client, destnode,
2546                                     tevent_timeval_zero(), &nodemap);
2547         if (ret != 0) {
2548                 return ret;
2549         }
2550
2551         flag_change.pnn = destnode;
2552         flag_change.old_flags = nodemap->node[destnode].flags;
2553         flag_change.new_flags = flag_change.old_flags | set;
2554         flag_change.new_flags &= ~clear;
2555
2556         count = list_of_connected_nodes(nodemap, -1, mem_ctx, &pnn_list);
2557         if (count == -1) {
2558                 return ENOMEM;
2559         }
2560
2561         ctdb_req_control_modify_flags(&request, &flag_change);
2562         ret = ctdb_client_control_multi(mem_ctx, ev, client, pnn_list, count,
2563                                         tevent_timeval_zero(), &request,
2564                                         NULL, NULL);
2565         return ret;
2566 }
2567
2568 struct ipreallocate_state {
2569         int status;
2570         bool done;
2571 };
2572
2573 static void ipreallocate_handler(uint64_t srvid, TDB_DATA data,
2574                                  void *private_data)
2575 {
2576         struct ipreallocate_state *state =
2577                 (struct ipreallocate_state *)private_data;
2578
2579         if (data.dsize != sizeof(int)) {
2580                 /* Ignore packet */
2581                 return;
2582         }
2583
2584         state->status = *(int *)data.dptr;
2585         state->done = true;
2586 }
2587
2588 static int ipreallocate(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb)
2589 {
2590         struct ctdb_srvid_message msg = { 0 };
2591         struct ipreallocate_state state;
2592         int ret;
2593
2594         msg.pnn = ctdb->pnn;
2595         msg.srvid = next_srvid(ctdb);
2596
2597         state.done = false;
2598         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
2599                                               msg.srvid,
2600                                               ipreallocate_handler, &state);
2601         if (ret != 0) {
2602                 return ret;
2603         }
2604
2605         while (true) {
2606                 ret = ctdb_message_takeover_run(mem_ctx, ctdb->ev,
2607                                                 ctdb->client,
2608                                                 CTDB_BROADCAST_CONNECTED,
2609                                                 &msg);
2610                 if (ret != 0) {
2611                         goto fail;
2612                 }
2613
2614                 ret = ctdb_client_wait_timeout(ctdb->ev, &state.done,
2615                                                TIMEOUT());
2616                 if (ret != 0) {
2617                         continue;
2618                 }
2619
2620                 if (state.status >= 0) {
2621                         ret = 0;
2622                 } else {
2623                         ret = state.status;
2624                 }
2625                 break;
2626         }
2627
2628 fail:
2629         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
2630                                            msg.srvid, &state);
2631         return ret;
2632 }
2633
2634 static int control_disable(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2635                            int argc, const char **argv)
2636 {
2637         int ret;
2638
2639         if (argc != 0) {
2640                 usage("disable");
2641         }
2642
2643         ret = check_flags(mem_ctx, ctdb, "disabled",
2644                           NODE_FLAGS_PERMANENTLY_DISABLED, true);
2645         if (ret == 0) {
2646                 return 0;
2647         }
2648
2649         ret = ctdb_ctrl_modflags(mem_ctx, ctdb->ev, ctdb->client,
2650                                  ctdb->cmd_pnn, TIMEOUT(),
2651                                  NODE_FLAGS_PERMANENTLY_DISABLED, 0);
2652         if (ret != 0) {
2653                 fprintf(stderr,
2654                         "Failed to set DISABLED flag on node %u\n",
2655                         ctdb->cmd_pnn);
2656                 return ret;
2657         }
2658
2659         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_PERMANENTLY_DISABLED, true);
2660         return ipreallocate(mem_ctx, ctdb);
2661 }
2662
2663 static int control_enable(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2664                           int argc, const char **argv)
2665 {
2666         int ret;
2667
2668         if (argc != 0) {
2669                 usage("enable");
2670         }
2671
2672         ret = check_flags(mem_ctx, ctdb, "disabled",
2673                           NODE_FLAGS_PERMANENTLY_DISABLED, false);
2674         if (ret == 0) {
2675                 return 0;
2676         }
2677
2678         ret = ctdb_ctrl_modflags(mem_ctx, ctdb->ev, ctdb->client,
2679                                  ctdb->cmd_pnn, TIMEOUT(),
2680                                  0, NODE_FLAGS_PERMANENTLY_DISABLED);
2681         if (ret != 0) {
2682                 fprintf(stderr, "Failed to reset DISABLED flag on node %u\n",
2683                         ctdb->cmd_pnn);
2684                 return ret;
2685         }
2686
2687         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_PERMANENTLY_DISABLED, false);
2688         return ipreallocate(mem_ctx, ctdb);
2689 }
2690
2691 static int control_stop(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2692                         int argc, const char **argv)
2693 {
2694         int ret;
2695
2696         if (argc != 0) {
2697                 usage("stop");
2698         }
2699
2700         ret = check_flags(mem_ctx, ctdb, "stopped",
2701                           NODE_FLAGS_STOPPED, true);
2702         if (ret == 0) {
2703                 return 0;
2704         }
2705
2706         ret = ctdb_ctrl_stop_node(mem_ctx, ctdb->ev, ctdb->client,
2707                                   ctdb->cmd_pnn, TIMEOUT());
2708         if (ret != 0) {
2709                 fprintf(stderr, "Failed to stop node %u\n", ctdb->cmd_pnn);
2710                 return ret;
2711         }
2712
2713         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_STOPPED, true);
2714         return ipreallocate(mem_ctx, ctdb);
2715 }
2716
2717 static int control_continue(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2718                             int argc, const char **argv)
2719 {
2720         int ret;
2721
2722         if (argc != 0) {
2723                 usage("continue");
2724         }
2725
2726         ret = check_flags(mem_ctx, ctdb, "stopped",
2727                           NODE_FLAGS_STOPPED, false);
2728         if (ret == 0) {
2729                 return 0;
2730         }
2731
2732         ret = ctdb_ctrl_continue_node(mem_ctx, ctdb->ev, ctdb->client,
2733                                       ctdb->cmd_pnn, TIMEOUT());
2734         if (ret != 0) {
2735                 fprintf(stderr, "Failed to continue stopped node %u\n",
2736                         ctdb->cmd_pnn);
2737                 return ret;
2738         }
2739
2740         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_STOPPED, false);
2741         return ipreallocate(mem_ctx, ctdb);
2742 }
2743
2744 static int control_ban(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2745                        int argc, const char **argv)
2746 {
2747         struct ctdb_ban_state ban_state;
2748         int ret;
2749
2750         if (argc != 1) {
2751                 usage("ban");
2752         }
2753
2754         ret = check_flags(mem_ctx, ctdb, "banned",
2755                           NODE_FLAGS_BANNED, true);
2756         if (ret == 0) {
2757                 return 0;
2758         }
2759
2760         ban_state.pnn = ctdb->cmd_pnn;
2761         ban_state.time = strtoul(argv[0], NULL, 0);
2762
2763         if (ban_state.time == 0) {
2764                 fprintf(stderr, "Ban time cannot be zero\n");
2765                 return EINVAL;
2766         }
2767
2768         ret = ctdb_ctrl_set_ban_state(mem_ctx, ctdb->ev, ctdb->client,
2769                                       ctdb->cmd_pnn, TIMEOUT(), &ban_state);
2770         if (ret != 0) {
2771                 fprintf(stderr, "Failed to ban node %u\n", ctdb->cmd_pnn);
2772                 return ret;
2773         }
2774
2775         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_BANNED, true);
2776         return ipreallocate(mem_ctx, ctdb);
2777
2778 }
2779
2780 static int control_unban(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2781                          int argc, const char **argv)
2782 {
2783         struct ctdb_ban_state ban_state;
2784         int ret;
2785
2786         if (argc != 0) {
2787                 usage("unban");
2788         }
2789
2790         ret = check_flags(mem_ctx, ctdb, "banned",
2791                           NODE_FLAGS_BANNED, false);
2792         if (ret == 0) {
2793                 return 0;
2794         }
2795
2796         ban_state.pnn = ctdb->cmd_pnn;
2797         ban_state.time = 0;
2798
2799         ret = ctdb_ctrl_set_ban_state(mem_ctx, ctdb->ev, ctdb->client,
2800                                       ctdb->cmd_pnn, TIMEOUT(), &ban_state);
2801         if (ret != 0) {
2802                 fprintf(stderr, "Failed to unban node %u\n", ctdb->cmd_pnn);
2803                 return ret;
2804         }
2805
2806         wait_for_flags(mem_ctx, ctdb, NODE_FLAGS_BANNED, false);
2807         return ipreallocate(mem_ctx, ctdb);
2808
2809 }
2810
2811 static int control_shutdown(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2812                             int argc, const char **argv)
2813 {
2814         int ret;
2815
2816         if (argc != 0) {
2817                 usage("shutdown");
2818         }
2819
2820         ret = ctdb_ctrl_shutdown(mem_ctx, ctdb->ev, ctdb->client,
2821                                  ctdb->cmd_pnn, TIMEOUT());
2822         if (ret != 0) {
2823                 fprintf(stderr, "Unable to shutdown node %u\n", ctdb->cmd_pnn);
2824                 return ret;
2825         }
2826
2827         return 0;
2828 }
2829
2830 static int get_generation(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2831                           uint32_t *generation)
2832 {
2833         uint32_t recmaster;
2834         int recmode;
2835         struct ctdb_vnn_map *vnnmap;
2836         int ret;
2837
2838 again:
2839         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
2840                                       ctdb->cmd_pnn, TIMEOUT(), &recmaster);
2841         if (ret != 0) {
2842                 fprintf(stderr, "Failed to find recovery master\n");
2843                 return ret;
2844         }
2845
2846         ret = ctdb_ctrl_get_recmode(mem_ctx, ctdb->ev, ctdb->client,
2847                                     recmaster, TIMEOUT(), &recmode);
2848         if (ret != 0) {
2849                 fprintf(stderr, "Failed to get recovery mode from node %u\n",
2850                         recmaster);
2851                 return ret;
2852         }
2853
2854         if (recmode == CTDB_RECOVERY_ACTIVE) {
2855                 sleep(1);
2856                 goto again;
2857         }
2858
2859         ret = ctdb_ctrl_getvnnmap(mem_ctx, ctdb->ev, ctdb->client,
2860                                   recmaster, TIMEOUT(), &vnnmap);
2861         if (ret != 0) {
2862                 fprintf(stderr, "Failed to get generation from node %u\n",
2863                         recmaster);
2864                 return ret;
2865         }
2866
2867         if (vnnmap->generation == INVALID_GENERATION) {
2868                 talloc_free(vnnmap);
2869                 sleep(1);
2870                 goto again;
2871         }
2872
2873         *generation = vnnmap->generation;
2874         talloc_free(vnnmap);
2875         return 0;
2876 }
2877
2878
2879 static int control_recover(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2880                            int argc, const char **argv)
2881 {
2882         uint32_t generation, next_generation;
2883         int ret;
2884
2885         if (argc != 0) {
2886                 usage("recover");
2887         }
2888
2889         ret = get_generation(mem_ctx, ctdb, &generation);
2890         if (ret != 0) {
2891                 return ret;
2892         }
2893
2894         ret = ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
2895                                     ctdb->cmd_pnn, TIMEOUT(),
2896                                     CTDB_RECOVERY_ACTIVE);
2897         if (ret != 0) {
2898                 fprintf(stderr, "Failed to set recovery mode active\n");
2899                 return ret;
2900         }
2901
2902         while (1) {
2903                 ret = get_generation(mem_ctx, ctdb, &next_generation);
2904                 if (ret != 0) {
2905                         fprintf(stderr,
2906                                 "Failed to confirm end of recovery\n");
2907                         return ret;
2908                 }
2909
2910                 if (next_generation != generation) {
2911                         break;
2912                 }
2913
2914                 sleep (1);
2915         }
2916
2917         return 0;
2918 }
2919
2920 static int control_ipreallocate(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2921                                 int argc, const char **argv)
2922 {
2923         if (argc != 0) {
2924                 usage("ipreallocate");
2925         }
2926
2927         return ipreallocate(mem_ctx, ctdb);
2928 }
2929
2930 static int control_isnotrecmaster(TALLOC_CTX *mem_ctx,
2931                                   struct ctdb_context *ctdb,
2932                                   int argc, const char **argv)
2933 {
2934         uint32_t recmaster;
2935         int ret;
2936
2937         if (argc != 0) {
2938                 usage("isnotrecmaster");
2939         }
2940
2941         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
2942                                       ctdb->pnn, TIMEOUT(), &recmaster);
2943         if (ret != 0) {
2944                 fprintf(stderr, "Failed to get recmaster\n");
2945                 return ret;
2946         }
2947
2948         if (recmaster != ctdb->pnn) {
2949                 printf("this node is not the recmaster\n");
2950                 return 1;
2951         }
2952
2953         printf("this node is the recmaster\n");
2954         return 0;
2955 }
2956
2957 static int control_gratarp(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2958                            int argc, const char **argv)
2959 {
2960         struct ctdb_addr_info addr_info;
2961         int ret;
2962
2963         if (argc != 2) {
2964                 usage("gratarp");
2965         }
2966
2967         if (! parse_ip(argv[0], NULL, 0, &addr_info.addr)) {
2968                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
2969                 return 1;
2970         }
2971         addr_info.iface = argv[1];
2972
2973         ret = ctdb_ctrl_send_gratuitous_arp(mem_ctx, ctdb->ev, ctdb->client,
2974                                             ctdb->cmd_pnn, TIMEOUT(),
2975                                             &addr_info);
2976         if (ret != 0) {
2977                 fprintf(stderr, "Unable to send gratuitous arp from node %u\n",
2978                         ctdb->cmd_pnn);
2979                 return ret;
2980         }
2981
2982         return 0;
2983 }
2984
2985 static int control_tickle(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
2986                            int argc, const char **argv)
2987 {
2988         ctdb_sock_addr src, dst;
2989         int ret;
2990
2991         if (argc != 0 && argc != 2) {
2992                 usage("tickle");
2993         }
2994
2995         if (argc == 0) {
2996                 struct ctdb_connection *clist;
2997                 int count;
2998                 int i, num_failed;
2999
3000                 ret = ctdb_parse_connections(stdin, mem_ctx, &count, &clist);
3001                 if (ret != 0) {
3002                         return ret;
3003                 }
3004
3005                 num_failed = 0;
3006                 for (i=0; i<count; i++) {
3007                         ret = ctdb_sys_send_tcp(&clist[i].src,
3008                                                 &clist[i].dst,
3009                                                 0, 0, 0);
3010                         if (ret != 0) {
3011                                 num_failed += 1;
3012                         }
3013                 }
3014
3015                 TALLOC_FREE(clist);
3016
3017                 if (num_failed > 0) {
3018                         fprintf(stderr, "Failed to send %d tickles\n",
3019                                 num_failed);
3020                         return 1;
3021                 }
3022
3023                 return 0;
3024         }
3025
3026
3027         if (! parse_ip_port(argv[0], &src)) {
3028                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3029                 return 1;
3030         }
3031
3032         if (! parse_ip_port(argv[1], &dst)) {
3033                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3034                 return 1;
3035         }
3036
3037         ret = ctdb_sys_send_tcp(&src, &dst, 0, 0, 0);
3038         if (ret != 0) {
3039                 fprintf(stderr, "Failed to send tickle ack\n");
3040                 return ret;
3041         }
3042
3043         return 0;
3044 }
3045
3046 static int control_gettickles(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3047                               int argc, const char **argv)
3048 {
3049         ctdb_sock_addr addr;
3050         struct ctdb_tickle_list *tickles;
3051         unsigned port = 0;
3052         int ret, i;
3053
3054         if (argc < 1 || argc > 2) {
3055                 usage("gettickles");
3056         }
3057
3058         if (argc == 2) {
3059                 port = strtoul(argv[1], NULL, 10);
3060         }
3061
3062         if (! parse_ip(argv[0], NULL, port, &addr)) {
3063                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3064                 return 1;
3065         }
3066
3067         ret = ctdb_ctrl_get_tcp_tickle_list(mem_ctx, ctdb->ev, ctdb->client,
3068                                             ctdb->cmd_pnn, TIMEOUT(), &addr,
3069                                             &tickles);
3070         if (ret != 0) {
3071                 fprintf(stderr, "Failed to get list of connections\n");
3072                 return ret;
3073         }
3074
3075         if (options.machinereadable) {
3076                 printf("%s%s%s%s%s%s%s%s%s\n",
3077                        options.sep,
3078                        "Source IP", options.sep,
3079                        "Port", options.sep,
3080                        "Destiation IP", options.sep,
3081                        "Port", options.sep);
3082                 for (i=0; i<tickles->num; i++) {
3083                         printf("%s%s%s%u%s%s%s%u%s\n", options.sep,
3084                                ctdb_sock_addr_to_string(
3085                                        mem_ctx, &tickles->conn[i].src, false),
3086                                options.sep,
3087                                ntohs(tickles->conn[i].src.ip.sin_port),
3088                                options.sep,
3089                                ctdb_sock_addr_to_string(
3090                                        mem_ctx, &tickles->conn[i].dst, false),
3091                                options.sep,
3092                                ntohs(tickles->conn[i].dst.ip.sin_port),
3093                                options.sep);
3094                 }
3095         } else {
3096                 printf("Connections for IP: %s\n",
3097                        ctdb_sock_addr_to_string(mem_ctx,
3098                                                 &tickles->addr, false));
3099                 printf("Num connections: %u\n", tickles->num);
3100                 for (i=0; i<tickles->num; i++) {
3101                         printf("SRC: %s   DST: %s\n",
3102                                ctdb_sock_addr_to_string(
3103                                        mem_ctx, &tickles->conn[i].src, true),
3104                                ctdb_sock_addr_to_string(
3105                                        mem_ctx, &tickles->conn[i].dst, true));
3106                 }
3107         }
3108
3109         talloc_free(tickles);
3110         return 0;
3111 }
3112
3113 typedef void (*clist_request_func)(struct ctdb_req_control *request,
3114                                    struct ctdb_connection *conn);
3115
3116 typedef int (*clist_reply_func)(struct ctdb_reply_control *reply);
3117
3118 struct process_clist_state {
3119         struct ctdb_connection *clist;
3120         int count;
3121         int num_failed, num_total;
3122         clist_reply_func reply_func;
3123 };
3124
3125 static void process_clist_done(struct tevent_req *subreq);
3126
3127 static struct tevent_req *process_clist_send(
3128                                         TALLOC_CTX *mem_ctx,
3129                                         struct ctdb_context *ctdb,
3130                                         struct ctdb_connection *clist,
3131                                         int count,
3132                                         clist_request_func request_func,
3133                                         clist_reply_func reply_func)
3134 {
3135         struct tevent_req *req, *subreq;
3136         struct process_clist_state *state;
3137         struct ctdb_req_control request;
3138         int i;
3139
3140         req = tevent_req_create(mem_ctx, &state, struct process_clist_state);
3141         if (req == NULL) {
3142                 return NULL;
3143         }
3144
3145         state->clist = clist;
3146         state->count = count;
3147         state->reply_func = reply_func;
3148
3149         for (i=0; i<count; i++) {
3150                 request_func(&request, &clist[i]);
3151                 subreq = ctdb_client_control_send(state, ctdb->ev,
3152                                                   ctdb->client, ctdb->cmd_pnn,
3153                                                   TIMEOUT(), &request);
3154                 if (tevent_req_nomem(subreq, req)) {
3155                         return tevent_req_post(req, ctdb->ev);
3156                 }
3157                 tevent_req_set_callback(subreq, process_clist_done, req);
3158         }
3159
3160         return req;
3161 }
3162
3163 static void process_clist_done(struct tevent_req *subreq)
3164 {
3165         struct tevent_req *req = tevent_req_callback_data(
3166                 subreq, struct tevent_req);
3167         struct process_clist_state *state = tevent_req_data(
3168                 req, struct process_clist_state);
3169         struct ctdb_reply_control *reply;
3170         int ret;
3171         bool status;
3172
3173         status = ctdb_client_control_recv(subreq, NULL, state, &reply);
3174         TALLOC_FREE(subreq);
3175         if (! status) {
3176                 state->num_failed += 1;
3177                 goto done;
3178         }
3179
3180         ret = state->reply_func(reply);
3181         if (ret != 0) {
3182                 state->num_failed += 1;
3183                 goto done;
3184         }
3185
3186 done:
3187         state->num_total += 1;
3188         if (state->num_total == state->count) {
3189                 tevent_req_done(req);
3190         }
3191 }
3192
3193 static int process_clist_recv(struct tevent_req *req)
3194 {
3195         struct process_clist_state *state = tevent_req_data(
3196                 req, struct process_clist_state);
3197
3198         return state->num_failed;
3199 }
3200
3201 static int control_addtickle(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3202                              int argc, const char **argv)
3203 {
3204         struct ctdb_connection conn;
3205         int ret;
3206
3207         if (argc != 0 && argc != 2) {
3208                 usage("addtickle");
3209         }
3210
3211         if (argc == 0) {
3212                 struct ctdb_connection *clist;
3213                 struct tevent_req *req;
3214                 int count;
3215
3216                 ret = ctdb_parse_connections(stdin, mem_ctx, &count, &clist);
3217                 if (ret != 0) {
3218                         return ret;
3219                 }
3220                 if (count == 0) {
3221                         return 0;
3222                 }
3223
3224                 req = process_clist_send(mem_ctx, ctdb, clist, count,
3225                                  ctdb_req_control_tcp_add_delayed_update,
3226                                  ctdb_reply_control_tcp_add_delayed_update);
3227                 if (req == NULL) {
3228                         talloc_free(clist);
3229                         return ENOMEM;
3230                 }
3231
3232                 tevent_req_poll(req, ctdb->ev);
3233                 talloc_free(clist);
3234
3235                 ret = process_clist_recv(req);
3236                 if (ret != 0) {
3237                         fprintf(stderr, "Failed to add %d tickles\n", ret);
3238                         return 1;
3239                 }
3240
3241                 return 0;
3242         }
3243
3244         if (! parse_ip_port(argv[0], &conn.src)) {
3245                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3246                 return 1;
3247         }
3248         if (! parse_ip_port(argv[1], &conn.dst)) {
3249                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3250                 return 1;
3251         }
3252
3253         ret = ctdb_ctrl_tcp_add_delayed_update(mem_ctx, ctdb->ev,
3254                                                ctdb->client, ctdb->cmd_pnn,
3255                                                TIMEOUT(), &conn);
3256         if (ret != 0) {
3257                 fprintf(stderr, "Failed to register connection\n");
3258                 return ret;
3259         }
3260
3261         return 0;
3262 }
3263
3264 static int control_deltickle(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3265                              int argc, const char **argv)
3266 {
3267         struct ctdb_connection conn;
3268         int ret;
3269
3270         if (argc != 0 && argc != 2) {
3271                 usage("deltickle");
3272         }
3273
3274         if (argc == 0) {
3275                 struct ctdb_connection *clist;
3276                 struct tevent_req *req;
3277                 int count;
3278
3279                 ret = ctdb_parse_connections(stdin, mem_ctx, &count, &clist);
3280                 if (ret != 0) {
3281                         return ret;
3282                 }
3283                 if (count == 0) {
3284                         return 0;
3285                 }
3286
3287                 req = process_clist_send(mem_ctx, ctdb, clist, count,
3288                                          ctdb_req_control_tcp_remove,
3289                                          ctdb_reply_control_tcp_remove);
3290                 if (req == NULL) {
3291                         talloc_free(clist);
3292                         return ENOMEM;
3293                 }
3294
3295                 tevent_req_poll(req, ctdb->ev);
3296                 talloc_free(clist);
3297
3298                 ret = process_clist_recv(req);
3299                 if (ret != 0) {
3300                         fprintf(stderr, "Failed to remove %d tickles\n", ret);
3301                         return 1;
3302                 }
3303
3304                 return 0;
3305         }
3306
3307         if (! parse_ip_port(argv[0], &conn.src)) {
3308                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3309                 return 1;
3310         }
3311         if (! parse_ip_port(argv[1], &conn.dst)) {
3312                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3313                 return 1;
3314         }
3315
3316         ret = ctdb_ctrl_tcp_remove(mem_ctx, ctdb->ev, ctdb->client,
3317                                    ctdb->cmd_pnn, TIMEOUT(), &conn);
3318         if (ret != 0) {
3319                 fprintf(stderr, "Failed to unregister connection\n");
3320                 return ret;
3321         }
3322
3323         return 0;
3324 }
3325
3326 static int control_listnodes(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3327                              int argc, const char **argv)
3328 {
3329         struct ctdb_node_map *nodemap;
3330         int i;
3331
3332         if (argc != 0) {
3333                 usage("listnodes");
3334         }
3335
3336         nodemap = read_nodes_file(mem_ctx, CTDB_UNKNOWN_PNN);
3337         if (nodemap == NULL) {
3338                 return 1;
3339         }
3340
3341         for (i=0; i<nodemap->num; i++) {
3342                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
3343                         continue;
3344                 }
3345
3346                 if (options.machinereadable) {
3347                         printf("%s%u%s%s%s\n", options.sep,
3348                                nodemap->node[i].pnn, options.sep,
3349                                ctdb_sock_addr_to_string(
3350                                        mem_ctx, &nodemap->node[i].addr, false),
3351                                options.sep);
3352                 } else {
3353                         printf("%s\n",
3354                                ctdb_sock_addr_to_string(
3355                                        mem_ctx, &nodemap->node[i].addr, false));
3356                 }
3357         }
3358
3359         return 0;
3360 }
3361
3362 static bool nodemap_identical(struct ctdb_node_map *nodemap1,
3363                               struct ctdb_node_map *nodemap2)
3364 {
3365         int i;
3366
3367         if (nodemap1->num != nodemap2->num) {
3368                 return false;
3369         }
3370
3371         for (i=0; i<nodemap1->num; i++) {
3372                 struct ctdb_node_and_flags *n1, *n2;
3373
3374                 n1 = &nodemap1->node[i];
3375                 n2 = &nodemap2->node[i];
3376
3377                 if ((n1->pnn != n2->pnn) ||
3378                     (n1->flags != n2->flags) ||
3379                     ! ctdb_sock_addr_same_ip(&n1->addr, &n2->addr)) {
3380                         return false;
3381                 }
3382         }
3383
3384         return true;
3385 }
3386
3387 static int check_node_file_changes(TALLOC_CTX *mem_ctx,
3388                                    struct ctdb_node_map *nm,
3389                                    struct ctdb_node_map *fnm,
3390                                    bool *reload)
3391 {
3392         int i;
3393         bool check_failed = false;
3394
3395         *reload = false;
3396
3397         for (i=0; i<nm->num; i++) {
3398                 if (i >= fnm->num) {
3399                         fprintf(stderr,
3400                                 "Node %u (%s) missing from nodes file\n",
3401                                 nm->node[i].pnn,
3402                                 ctdb_sock_addr_to_string(
3403                                         mem_ctx, &nm->node[i].addr, false));
3404                         check_failed = true;
3405                         continue;
3406                 }
3407                 if (nm->node[i].flags & NODE_FLAGS_DELETED &&
3408                     fnm->node[i].flags & NODE_FLAGS_DELETED) {
3409                         /* Node remains deleted */
3410                         continue;
3411                 }
3412
3413                 if (! (nm->node[i].flags & NODE_FLAGS_DELETED) &&
3414                     ! (fnm->node[i].flags & NODE_FLAGS_DELETED)) {
3415                         /* Node not newly nor previously deleted */
3416                         if (! ctdb_same_ip(&nm->node[i].addr,
3417                                            &fnm->node[i].addr)) {
3418                                 fprintf(stderr,
3419                                         "Node %u has changed IP address"
3420                                         " (was %s, now %s)\n",
3421                                         nm->node[i].pnn,
3422                                         ctdb_sock_addr_to_string(
3423                                                 mem_ctx,
3424                                                 &nm->node[i].addr, false),
3425                                         ctdb_sock_addr_to_string(
3426                                                 mem_ctx,
3427                                                 &fnm->node[i].addr, false));
3428                                 check_failed = true;
3429                         } else {
3430                                 if (nm->node[i].flags & NODE_FLAGS_DISCONNECTED) {
3431                                         fprintf(stderr,
3432                                                 "WARNING: Node %u is disconnected."
3433                                                 " You MUST fix this node manually!\n",
3434                                                 nm->node[i].pnn);
3435                                 }
3436                         }
3437                         continue;
3438                 }
3439
3440                 if (fnm->node[i].flags & NODE_FLAGS_DELETED) {
3441                         /* Node is being deleted */
3442                         printf("Node %u is DELETED\n", nm->node[i].pnn);
3443                         *reload = true;
3444                         if (! (nm->node[i].flags & NODE_FLAGS_DISCONNECTED)) {
3445                                 fprintf(stderr,
3446                                         "ERROR: Node %u is still connected\n",
3447                                         nm->node[i].pnn);
3448                                 check_failed = true;
3449                         }
3450                         continue;
3451                 }
3452
3453                 if (nm->node[i].flags & NODE_FLAGS_DELETED) {
3454                         /* Node was previously deleted */
3455                         printf("Node %u is UNDELETED\n", nm->node[i].pnn);
3456                         *reload = true;
3457                 }
3458         }
3459
3460         if (check_failed) {
3461                 fprintf(stderr,
3462                         "ERROR: Nodes will not be reloaded due to previous error\n");
3463                 return 1;
3464         }
3465
3466         /* Leftover nodes in file are NEW */
3467         for (; i < fnm->num; i++) {
3468                 printf("Node %u is NEW\n", fnm->node[i].pnn);
3469                 *reload = true;
3470         }
3471
3472         return 0;
3473 }
3474
3475 struct disable_recoveries_state {
3476         uint32_t *pnn_list;
3477         int node_count;
3478         bool *reply;
3479         int status;
3480         bool done;
3481 };
3482
3483 static void disable_recoveries_handler(uint64_t srvid, TDB_DATA data,
3484                                        void *private_data)
3485 {
3486         struct disable_recoveries_state *state =
3487                 (struct disable_recoveries_state *)private_data;
3488         int ret, i;
3489
3490         if (data.dsize != sizeof(int)) {
3491                 /* Ignore packet */
3492                 return;
3493         }
3494
3495         /* ret will be a PNN (i.e. >=0) on success, or negative on error */
3496         ret = *(int *)data.dptr;
3497         if (ret < 0) {
3498                 state->status = ret;
3499                 state->done = true;
3500                 return;
3501         }
3502         for (i=0; i<state->node_count; i++) {
3503                 if (state->pnn_list[i] == ret) {
3504                         state->reply[i] = true;
3505                         break;
3506                 }
3507         }
3508
3509         state->done = true;
3510         for (i=0; i<state->node_count; i++) {
3511                 if (! state->reply[i]) {
3512                         state->done = false;
3513                         break;
3514                 }
3515         }
3516 }
3517
3518 static int disable_recoveries(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3519                               uint32_t timeout, uint32_t *pnn_list, int count)
3520 {
3521         struct ctdb_disable_message disable = { 0 };
3522         struct disable_recoveries_state state;
3523         int ret, i;
3524
3525         disable.pnn = ctdb->pnn;
3526         disable.srvid = next_srvid(ctdb);
3527         disable.timeout = timeout;
3528
3529         state.pnn_list = pnn_list;
3530         state.node_count = count;
3531         state.done = false;
3532         state.status = 0;
3533         state.reply = talloc_zero_array(mem_ctx, bool, count);
3534         if (state.reply == NULL) {
3535                 return ENOMEM;
3536         }
3537
3538         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
3539                                               disable.srvid,
3540                                               disable_recoveries_handler,
3541                                               &state);
3542         if (ret != 0) {
3543                 return ret;
3544         }
3545
3546         for (i=0; i<count; i++) {
3547                 ret = ctdb_message_disable_recoveries(mem_ctx, ctdb->ev,
3548                                                       ctdb->client,
3549                                                       pnn_list[i],
3550                                                       &disable);
3551                 if (ret != 0) {
3552                         goto fail;
3553                 }
3554         }
3555
3556         ret = ctdb_client_wait_timeout(ctdb->ev, &state.done, TIMEOUT());
3557         if (ret == ETIME) {
3558                 fprintf(stderr, "Timed out waiting to disable recoveries\n");
3559         } else {
3560                 ret = (state.status >= 0 ? 0 : 1);
3561         }
3562
3563 fail:
3564         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
3565                                            disable.srvid, &state);
3566         return ret;
3567 }
3568
3569 static int control_reloadnodes(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3570                                int argc, const char **argv)
3571 {
3572         struct ctdb_node_map *nodemap = NULL;
3573         struct ctdb_node_map *file_nodemap;
3574         struct ctdb_node_map *remote_nodemap;
3575         struct ctdb_req_control request;
3576         struct ctdb_reply_control **reply;
3577         bool reload;
3578         int ret, i;
3579         uint32_t *pnn_list;
3580         int count;
3581
3582         nodemap = get_nodemap(ctdb, false);
3583         if (nodemap == NULL) {
3584                 return 1;
3585         }
3586
3587         file_nodemap = read_nodes_file(mem_ctx, ctdb->pnn);
3588         if (file_nodemap == NULL) {
3589                 return 1;
3590         }
3591
3592         for (i=0; i<nodemap->num; i++) {
3593                 if (nodemap->node[i].flags & NODE_FLAGS_DISCONNECTED) {
3594                         continue;
3595                 }
3596
3597                 ret = ctdb_ctrl_get_nodes_file(mem_ctx, ctdb->ev, ctdb->client,
3598                                                nodemap->node[i].pnn, TIMEOUT(),
3599                                                &remote_nodemap);
3600                 if (ret != 0) {
3601                         fprintf(stderr,
3602                                 "ERROR: Failed to get nodes file from node %u\n",
3603                                 nodemap->node[i].pnn);
3604                         return ret;
3605                 }
3606
3607                 if (! nodemap_identical(file_nodemap, remote_nodemap)) {
3608                         fprintf(stderr,
3609                                 "ERROR: Nodes file on node %u differs"
3610                                  " from current node (%u)\n",
3611                                  nodemap->node[i].pnn, ctdb->pnn);
3612                         return 1;
3613                 }
3614         }
3615
3616         ret = check_node_file_changes(mem_ctx, nodemap, file_nodemap, &reload);
3617         if (ret != 0) {
3618                 return ret;
3619         }
3620
3621         if (! reload) {
3622                 fprintf(stderr, "No change in nodes file,"
3623                                 " skipping unnecessary reload\n");
3624                 return 0;
3625         }
3626
3627         count = list_of_connected_nodes(nodemap, CTDB_UNKNOWN_PNN,
3628                                         mem_ctx, &pnn_list);
3629         if (count <= 0) {
3630                 fprintf(stderr, "Memory allocation error\n");
3631                 return 1;
3632         }
3633
3634         ret = disable_recoveries(mem_ctx, ctdb, 2*options.timelimit,
3635                                  pnn_list, count);
3636         if (ret != 0) {
3637                 fprintf(stderr, "Failed to disable recoveries\n");
3638                 return ret;
3639         }
3640
3641         ctdb_req_control_reload_nodes_file(&request);
3642         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
3643                                         pnn_list, count, TIMEOUT(),
3644                                         &request, NULL, &reply);
3645         if (ret != 0) {
3646                 bool failed = false;
3647
3648                 for (i=0; i<count; i++) {
3649                         ret = ctdb_reply_control_reload_nodes_file(reply[i]);
3650                         if (ret != 0) {
3651                                 fprintf(stderr,
3652                                         "Node %u failed to reload nodes\n",
3653                                         pnn_list[i]);
3654                                 failed = true;
3655                         }
3656                 }
3657                 if (failed) {
3658                         fprintf(stderr,
3659                                 "You MUST fix failed nodes manually!\n");
3660                 }
3661         }
3662
3663         ret = disable_recoveries(mem_ctx, ctdb, 0, pnn_list, count);
3664         if (ret != 0) {
3665                 fprintf(stderr, "Failed to enable recoveries\n");
3666                 return ret;
3667         }
3668
3669         return 0;
3670 }
3671
3672 static int moveip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3673                   ctdb_sock_addr *addr, uint32_t pnn)
3674 {
3675         struct ctdb_public_ip_list *pubip_list;
3676         struct ctdb_public_ip pubip;
3677         struct ctdb_node_map *nodemap;
3678         struct ctdb_req_control request;
3679         uint32_t *pnn_list;
3680         int ret, i, count;
3681
3682         ret = ctdb_message_disable_ip_check(mem_ctx, ctdb->ev, ctdb->client,
3683                                             CTDB_BROADCAST_CONNECTED,
3684                                             2*options.timelimit);
3685         if (ret != 0) {
3686                 fprintf(stderr, "Failed to disable IP check\n");
3687                 return ret;
3688         }
3689
3690         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3691                                        pnn, TIMEOUT(), false, &pubip_list);
3692         if (ret != 0) {
3693                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3694                         pnn);
3695                 return ret;
3696         }
3697
3698         for (i=0; i<pubip_list->num; i++) {
3699                 if (ctdb_same_ip(addr, &pubip_list->ip[i].addr)) {
3700                         break;
3701                 }
3702         }
3703
3704         if (i == pubip_list->num) {
3705                 fprintf(stderr, "Node %u CANNOT host IP address %s\n",
3706                         pnn, ctdb_sock_addr_to_string(mem_ctx, addr, false));
3707                 return 1;
3708         }
3709
3710         nodemap = get_nodemap(ctdb, false);
3711         if (nodemap == NULL) {
3712                 return 1;
3713         }
3714
3715         count = list_of_active_nodes(nodemap, pnn, mem_ctx, &pnn_list);
3716         if (count <= 0) {
3717                 fprintf(stderr, "Memory allocation error\n");
3718                 return 1;
3719         }
3720
3721         pubip.pnn = pnn;
3722         pubip.addr = *addr;
3723         ctdb_req_control_release_ip(&request, &pubip);
3724
3725         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
3726                                         pnn_list, count, TIMEOUT(),
3727                                         &request, NULL, NULL);
3728         if (ret != 0) {
3729                 fprintf(stderr, "Failed to release IP on nodes\n");
3730                 return ret;
3731         }
3732
3733         ret = ctdb_ctrl_takeover_ip(mem_ctx, ctdb->ev, ctdb->client,
3734                                     pnn, TIMEOUT(), &pubip);
3735         if (ret != 0) {
3736                 fprintf(stderr, "Failed to takeover IP on node %u\n", pnn);
3737                 return ret;
3738         }
3739
3740         return 0;
3741 }
3742
3743 static int control_moveip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3744                           int argc, const char **argv)
3745 {
3746         ctdb_sock_addr addr;
3747         uint32_t pnn;
3748         int ret, retries = 0;
3749
3750         if (argc != 2) {
3751                 usage("moveip");
3752         }
3753
3754         if (! parse_ip(argv[0], NULL, 0, &addr)) {
3755                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3756                 return 1;
3757         }
3758
3759         pnn = strtoul(argv[1], NULL, 10);
3760         if (pnn == CTDB_UNKNOWN_PNN) {
3761                 fprintf(stderr, "Invalid PNN %s\n", argv[1]);
3762                 return 1;
3763         }
3764
3765         while (retries < 5) {
3766                 ret = moveip(mem_ctx, ctdb, &addr, pnn);
3767                 if (ret == 0) {
3768                         break;
3769                 }
3770
3771                 sleep(3);
3772                 retries++;
3773         }
3774
3775         if (ret != 0) {
3776                 fprintf(stderr, "Failed to move IP %s to node %u\n",
3777                         argv[0], pnn);
3778                 return ret;
3779         }
3780
3781         return 0;
3782 }
3783
3784 static int rebalancenode(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3785                          uint32_t pnn)
3786 {
3787         int ret;
3788
3789         ret = ctdb_message_rebalance_node(mem_ctx, ctdb->ev, ctdb->client,
3790                                           CTDB_BROADCAST_CONNECTED, pnn);
3791         if (ret != 0) {
3792                 fprintf(stderr,
3793                         "Failed to ask recovery master to distribute IPs\n");
3794                 return ret;
3795         }
3796
3797         return 0;
3798 }
3799
3800 static int control_addip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3801                          int argc, const char **argv)
3802 {
3803         ctdb_sock_addr addr;
3804         struct ctdb_public_ip_list *pubip_list;
3805         struct ctdb_addr_info addr_info;
3806         unsigned int mask;
3807         int ret, i, retries = 0;
3808
3809         if (argc != 2) {
3810                 usage("addip");
3811         }
3812
3813         if (! parse_ip_mask(argv[0], argv[1], &addr, &mask)) {
3814                 fprintf(stderr, "Invalid IP/Mask %s\n", argv[0]);
3815                 return 1;
3816         }
3817
3818         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3819                                        ctdb->cmd_pnn, TIMEOUT(),
3820                                        false, &pubip_list);
3821         if (ret != 0) {
3822                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3823                         ctdb->cmd_pnn);
3824                 return 1;
3825         }
3826
3827         for (i=0; i<pubip_list->num; i++) {
3828                 if (ctdb_same_ip(&addr, &pubip_list->ip[i].addr)) {
3829                         fprintf(stderr, "Node already knows about IP %s\n",
3830                                 ctdb_sock_addr_to_string(mem_ctx,
3831                                                          &addr, false));
3832                         return 0;
3833                 }
3834         }
3835
3836         addr_info.addr = addr;
3837         addr_info.mask = mask;
3838         addr_info.iface = argv[1];
3839
3840         while (retries < 5) {
3841                 ret = ctdb_ctrl_add_public_ip(mem_ctx, ctdb->ev, ctdb->client,
3842                                               ctdb->cmd_pnn, TIMEOUT(),
3843                                               &addr_info);
3844                 if (ret == 0) {
3845                         break;
3846                 }
3847
3848                 sleep(3);
3849                 retries++;
3850         }
3851
3852         if (ret != 0) {
3853                 fprintf(stderr, "Failed to add public IP to node %u."
3854                                 " Giving up\n", ctdb->cmd_pnn);
3855                 return ret;
3856         }
3857
3858         ret = rebalancenode(mem_ctx, ctdb, ctdb->cmd_pnn);
3859         if (ret != 0) {
3860                 return ret;
3861         }
3862
3863         return 0;
3864 }
3865
3866 static int control_delip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3867                          int argc, const char **argv)
3868 {
3869         ctdb_sock_addr addr;
3870         struct ctdb_public_ip_list *pubip_list;
3871         struct ctdb_addr_info addr_info;
3872         int ret, i;
3873
3874         if (argc != 1) {
3875                 usage("delip");
3876         }
3877
3878         if (! parse_ip(argv[0], NULL, 0, &addr)) {
3879                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3880                 return 1;
3881         }
3882
3883         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3884                                        ctdb->cmd_pnn, TIMEOUT(),
3885                                        false, &pubip_list);
3886         if (ret != 0) {
3887                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3888                         ctdb->cmd_pnn);
3889                 return 1;
3890         }
3891
3892         for (i=0; i<pubip_list->num; i++) {
3893                 if (ctdb_same_ip(&addr, &pubip_list->ip[i].addr)) {
3894                         break;
3895                 }
3896         }
3897
3898         if (i == pubip_list->num) {
3899                 fprintf(stderr, "Node does not know about IP address %s\n",
3900                         ctdb_sock_addr_to_string(mem_ctx, &addr, false));
3901                 return 0;
3902         }
3903
3904         addr_info.addr = addr;
3905         addr_info.mask = 0;
3906         addr_info.iface = NULL;
3907
3908         ret = ctdb_ctrl_del_public_ip(mem_ctx, ctdb->ev, ctdb->client,
3909                                       ctdb->cmd_pnn, TIMEOUT(), &addr_info);
3910         if (ret != 0) {
3911                 fprintf(stderr, "Failed to delete public IP from node %u\n",
3912                         ctdb->cmd_pnn);
3913                 return ret;
3914         }
3915
3916         return 0;
3917 }
3918
3919 #define DB_VERSION      3
3920 #define MAX_DB_NAME     64
3921 #define MAX_REC_BUFFER_SIZE     (100*1000)
3922
3923 struct db_header {
3924         unsigned long version;
3925         time_t timestamp;
3926         unsigned long flags;
3927         unsigned long nbuf;
3928         unsigned long nrec;
3929         char name[MAX_DB_NAME];
3930 };
3931
3932 struct backup_state {
3933         TALLOC_CTX *mem_ctx;
3934         struct ctdb_rec_buffer *recbuf;
3935         uint32_t db_id;
3936         int fd;
3937         unsigned int nbuf, nrec;
3938 };
3939
3940 static int backup_handler(uint32_t reqid, struct ctdb_ltdb_header *header,
3941                           TDB_DATA key, TDB_DATA data, void *private_data)
3942 {
3943         struct backup_state *state = (struct backup_state *)private_data;
3944         size_t len;
3945         int ret;
3946
3947         if (state->recbuf == NULL) {
3948                 state->recbuf = ctdb_rec_buffer_init(state->mem_ctx,
3949                                                      state->db_id);
3950                 if (state->recbuf == NULL) {
3951                         return ENOMEM;
3952                 }
3953         }
3954
3955         ret = ctdb_rec_buffer_add(state->recbuf, state->recbuf, reqid,
3956                                   header, key, data);
3957         if (ret != 0) {
3958                 return ret;
3959         }
3960
3961         len = ctdb_rec_buffer_len(state->recbuf);
3962         if (len < MAX_REC_BUFFER_SIZE) {
3963                 return 0;
3964         }
3965
3966         ret = ctdb_rec_buffer_write(state->recbuf, state->fd);
3967         if (ret != 0) {
3968                 fprintf(stderr, "Failed to write records to backup file\n");
3969                 return ret;
3970         }
3971
3972         state->nbuf += 1;
3973         state->nrec += state->recbuf->count;
3974         TALLOC_FREE(state->recbuf);
3975
3976         return 0;
3977 }
3978
3979 static int control_backupdb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3980                             int argc, const char **argv)
3981 {
3982         const char *db_name;
3983         struct ctdb_db_context *db;
3984         uint32_t db_id;
3985         uint8_t db_flags;
3986         struct backup_state state;
3987         struct db_header db_hdr;
3988         int fd, ret;
3989
3990         if (argc != 2) {
3991                 usage("backupdb");
3992         }
3993
3994         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
3995                 return 1;
3996         }
3997
3998         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
3999                           db_flags, &db);
4000         if (ret != 0) {
4001                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4002                 return ret;
4003         }
4004
4005         fd = open(argv[1], O_RDWR|O_CREAT, 0600);
4006         if (fd == -1) {
4007                 ret = errno;
4008                 fprintf(stderr, "Failed to open file %s for writing\n",
4009                         argv[1]);
4010                 return ret;
4011         }
4012
4013         /* Write empty header first */
4014         ZERO_STRUCT(db_hdr);
4015         ret = write(fd, &db_hdr, sizeof(struct db_header));
4016         if (ret == -1) {
4017                 ret = errno;
4018                 close(fd);
4019                 fprintf(stderr, "Failed to write header to file %s\n", argv[1]);
4020                 return ret;
4021         }
4022
4023         state.mem_ctx = mem_ctx;
4024         state.recbuf = NULL;
4025         state.fd = fd;
4026         state.nbuf = 0;
4027         state.nrec = 0;
4028
4029         ret = ctdb_db_traverse_local(db, true, false, backup_handler, &state);
4030         if (ret != 0) {
4031                 fprintf(stderr, "Failed to collect records from DB %s\n",
4032                         db_name);
4033                 close(fd);
4034                 return ret;
4035         }
4036
4037         if (state.recbuf != NULL) {
4038                 ret = ctdb_rec_buffer_write(state.recbuf, state.fd);
4039                 if (ret != 0) {
4040                         fprintf(stderr,
4041                                 "Failed to write records to backup file\n");
4042                         close(fd);
4043                         return ret;
4044                 }
4045
4046                 state.nbuf += 1;
4047                 state.nrec += state.recbuf->count;
4048                 TALLOC_FREE(state.recbuf);
4049         }
4050
4051         db_hdr.version = DB_VERSION;
4052         db_hdr.timestamp = time(NULL);
4053         db_hdr.flags = db_flags;
4054         db_hdr.nbuf = state.nbuf;
4055         db_hdr.nrec = state.nrec;
4056         strncpy(db_hdr.name, db_name, MAX_DB_NAME-1);
4057
4058         lseek(fd, 0, SEEK_SET);
4059         ret = write(fd, &db_hdr, sizeof(struct db_header));
4060         if (ret == -1) {
4061                 ret = errno;
4062                 close(fd);
4063                 fprintf(stderr, "Failed to write header to file %s\n", argv[1]);
4064                 return ret;
4065         }
4066
4067         close(fd);
4068         printf("Database backed up to %s\n", argv[1]);
4069         return 0;
4070 }
4071
4072 static int control_restoredb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4073                              int argc, const char **argv)
4074 {
4075         const char *db_name = NULL;
4076         struct ctdb_db_context *db;
4077         struct db_header db_hdr;
4078         struct ctdb_node_map *nodemap;
4079         struct ctdb_req_control request;
4080         struct ctdb_reply_control **reply;
4081         struct ctdb_transdb wipedb;
4082         struct ctdb_pulldb_ext pulldb;
4083         struct ctdb_rec_buffer *recbuf;
4084         uint32_t generation;
4085         uint32_t *pnn_list;
4086         char timebuf[128];
4087         ssize_t n;
4088         int fd, i;
4089         int count, ret;
4090         uint8_t db_flags;
4091
4092         if (argc < 1 || argc > 2) {
4093                 usage("restoredb");
4094         }
4095
4096         fd = open(argv[0], O_RDONLY, 0600);
4097         if (fd == -1) {
4098                 ret = errno;
4099                 fprintf(stderr, "Failed to open file %s for reading\n",
4100                         argv[0]);
4101                 return ret;
4102         }
4103
4104         if (argc == 2) {
4105                 db_name = argv[1];
4106         }
4107
4108         n = read(fd, &db_hdr, sizeof(struct db_header));
4109         if (n == -1) {
4110                 ret = errno;
4111                 close(fd);
4112                 fprintf(stderr, "Failed to read db header from file %s\n",
4113                         argv[0]);
4114                 return ret;
4115         }
4116         db_hdr.name[sizeof(db_hdr.name)-1] = '\0';
4117
4118         if (db_hdr.version != DB_VERSION) {
4119                 fprintf(stderr,
4120                         "Wrong version of backup file, expected %u, got %lu\n",
4121                         DB_VERSION, db_hdr.version);
4122                 close(fd);
4123                 return EINVAL;
4124         }
4125
4126         if (db_name == NULL) {
4127                 db_name = db_hdr.name;
4128         }
4129
4130         strftime(timebuf, sizeof(timebuf)-1, "%Y/%m/%d %H:%M:%S",
4131                  localtime(&db_hdr.timestamp));
4132         printf("Restoring database %s from backup @ %s\n", db_name, timebuf);
4133
4134         db_flags = db_hdr.flags & 0xff;
4135         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4136                           db_flags, &db);
4137         if (ret != 0) {
4138                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4139                 close(fd);
4140                 return ret;
4141         }
4142
4143         nodemap = get_nodemap(ctdb, false);
4144         if (nodemap == NULL) {
4145                 fprintf(stderr, "Failed to get nodemap\n");
4146                 close(fd);
4147                 return ENOMEM;
4148         }
4149
4150         ret = get_generation(mem_ctx, ctdb, &generation);
4151         if (ret != 0) {
4152                 fprintf(stderr, "Failed to get current generation\n");
4153                 close(fd);
4154                 return ret;
4155         }
4156
4157         count = list_of_active_nodes(nodemap, CTDB_UNKNOWN_PNN, mem_ctx,
4158                                      &pnn_list);
4159         if (count <= 0) {
4160                 close(fd);
4161                 return ENOMEM;
4162         }
4163
4164         wipedb.db_id = ctdb_db_id(db);
4165         wipedb.tid = generation;
4166
4167         ctdb_req_control_db_freeze(&request, wipedb.db_id);
4168         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4169                                         ctdb->client, pnn_list, count,
4170                                         TIMEOUT(), &request, NULL, NULL);
4171         if (ret != 0) {
4172                 goto failed;
4173         }
4174
4175
4176         ctdb_req_control_db_transaction_start(&request, &wipedb);
4177         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4178                                         pnn_list, count, TIMEOUT(),
4179                                         &request, NULL, NULL);
4180         if (ret != 0) {
4181                 goto failed;
4182         }
4183
4184         ctdb_req_control_wipe_database(&request, &wipedb);
4185         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4186                                         pnn_list, count, TIMEOUT(),
4187                                         &request, NULL, NULL);
4188         if (ret != 0) {
4189                 goto failed;
4190         }
4191
4192         pulldb.db_id = ctdb_db_id(db);
4193         pulldb.lmaster = 0;
4194         pulldb.srvid = SRVID_CTDB_PUSHDB;
4195
4196         ctdb_req_control_db_push_start(&request, &pulldb);
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         for (i=0; i<db_hdr.nbuf; i++) {
4205                 struct ctdb_req_message message;
4206                 TDB_DATA data;
4207                 size_t np;
4208
4209                 ret = ctdb_rec_buffer_read(fd, mem_ctx, &recbuf);
4210                 if (ret != 0) {
4211                         goto failed;
4212                 }
4213
4214                 data.dsize = ctdb_rec_buffer_len(recbuf);
4215                 data.dptr = talloc_size(mem_ctx, data.dsize);
4216                 if (data.dptr == NULL) {
4217                         goto failed;
4218                 }
4219
4220                 ctdb_rec_buffer_push(recbuf, data.dptr, &np);
4221
4222                 message.srvid = pulldb.srvid;
4223                 message.data.data = data;
4224
4225                 ret = ctdb_client_message_multi(mem_ctx, ctdb->ev,
4226                                                 ctdb->client,
4227                                                 pnn_list, count,
4228                                                 &message, NULL);
4229                 if (ret != 0) {
4230                         goto failed;
4231                 }
4232
4233                 talloc_free(recbuf);
4234                 talloc_free(data.dptr);
4235         }
4236
4237         ctdb_req_control_db_push_confirm(&request, pulldb.db_id);
4238         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4239                                         pnn_list, count, TIMEOUT(),
4240                                         &request, NULL, &reply);
4241         if (ret != 0) {
4242                 goto failed;
4243         }
4244
4245         for (i=0; i<count; i++) {
4246                 uint32_t num_records;
4247
4248                 ret = ctdb_reply_control_db_push_confirm(reply[i],
4249                                                          &num_records);
4250                 if (ret != 0) {
4251                         fprintf(stderr, "Invalid response from node %u\n",
4252                                 pnn_list[i]);
4253                         goto failed;
4254                 }
4255
4256                 if (num_records != db_hdr.nrec) {
4257                         fprintf(stderr, "Node %u received %u of %lu records\n",
4258                                 pnn_list[i], num_records, db_hdr.nrec);
4259                         goto failed;
4260                 }
4261         }
4262
4263         ctdb_req_control_db_set_healthy(&request, wipedb.db_id);
4264         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4265                                         pnn_list, count, TIMEOUT(),
4266                                         &request, NULL, NULL);
4267         if (ret != 0) {
4268                 goto failed;
4269         }
4270
4271         ctdb_req_control_db_transaction_commit(&request, &wipedb);
4272         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4273                                         pnn_list, count, TIMEOUT(),
4274                                         &request, NULL, NULL);
4275         if (ret != 0) {
4276                 goto failed;
4277         }
4278
4279         ctdb_req_control_db_thaw(&request, wipedb.db_id);
4280         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4281                                         ctdb->client, pnn_list, count,
4282                                         TIMEOUT(), &request, NULL, NULL);
4283         if (ret != 0) {
4284                 goto failed;
4285         }
4286
4287         printf("Database %s restored\n", db_name);
4288         close(fd);
4289         return 0;
4290
4291
4292 failed:
4293         close(fd);
4294         ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
4295                               ctdb->pnn, TIMEOUT(), CTDB_RECOVERY_ACTIVE);
4296         return ret;
4297 }
4298
4299 struct dumpdbbackup_state {
4300         ctdb_rec_parser_func_t parser;
4301         struct dump_record_state sub_state;
4302 };
4303
4304 static int dumpdbbackup_handler(uint32_t reqid,
4305                                 struct ctdb_ltdb_header *header,
4306                                 TDB_DATA key, TDB_DATA data,
4307                                 void *private_data)
4308 {
4309         struct dumpdbbackup_state *state =
4310                 (struct dumpdbbackup_state *)private_data;
4311         struct ctdb_ltdb_header hdr;
4312         int ret;
4313
4314         ret = ctdb_ltdb_header_extract(&data, &hdr);
4315         if (ret != 0) {
4316                 return ret;
4317         }
4318
4319         return state->parser(reqid, &hdr, key, data, &state->sub_state);
4320 }
4321
4322 static int control_dumpdbbackup(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4323                                 int argc, const char **argv)
4324 {
4325         struct db_header db_hdr;
4326         char timebuf[128];
4327         struct dumpdbbackup_state state;
4328         ssize_t n;
4329         int fd, ret, i;
4330
4331         if (argc != 1) {
4332                 usage("dumpbackup");
4333         }
4334
4335         fd = open(argv[0], O_RDONLY, 0600);
4336         if (fd == -1) {
4337                 ret = errno;
4338                 fprintf(stderr, "Failed to open file %s for reading\n",
4339                         argv[0]);
4340                 return ret;
4341         }
4342
4343         n = read(fd, &db_hdr, sizeof(struct db_header));
4344         if (n == -1) {
4345                 ret = errno;
4346                 close(fd);
4347                 fprintf(stderr, "Failed to read db header from file %s\n",
4348                         argv[0]);
4349                 return ret;
4350         }
4351         db_hdr.name[sizeof(db_hdr.name)-1] = '\0';
4352
4353         if (db_hdr.version != DB_VERSION) {
4354                 fprintf(stderr,
4355                         "Wrong version of backup file, expected %u, got %lu\n",
4356                         DB_VERSION, db_hdr.version);
4357                 close(fd);
4358                 return EINVAL;
4359         }
4360
4361         strftime(timebuf, sizeof(timebuf)-1, "%Y/%m/%d %H:%M:%S",
4362                  localtime(&db_hdr.timestamp));
4363         printf("Dumping database %s from backup @ %s\n",
4364                db_hdr.name, timebuf);
4365
4366         state.parser = dump_record;
4367         state.sub_state.count = 0;
4368
4369         for (i=0; i<db_hdr.nbuf; i++) {
4370                 struct ctdb_rec_buffer *recbuf;
4371
4372                 ret = ctdb_rec_buffer_read(fd, mem_ctx, &recbuf);
4373                 if (ret != 0) {
4374                         fprintf(stderr, "Failed to read records\n");
4375                         close(fd);
4376                         return ret;
4377                 }
4378
4379                 ret = ctdb_rec_buffer_traverse(recbuf, dumpdbbackup_handler,
4380                                                &state);
4381                 if (ret != 0) {
4382                         fprintf(stderr, "Failed to dump records\n");
4383                         close(fd);
4384                         return ret;
4385                 }
4386         }
4387
4388         close(fd);
4389         printf("Dumped %u record(s)\n", state.sub_state.count);
4390         return 0;
4391 }
4392
4393 static int control_wipedb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4394                           int argc, const char **argv)
4395 {
4396         const char *db_name;
4397         struct ctdb_db_context *db;
4398         uint32_t db_id;
4399         uint8_t db_flags;
4400         struct ctdb_node_map *nodemap;
4401         struct ctdb_req_control request;
4402         struct ctdb_transdb wipedb;
4403         uint32_t generation;
4404         uint32_t *pnn_list;
4405         int count, ret;
4406
4407         if (argc != 1) {
4408                 usage("wipedb");
4409         }
4410
4411         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
4412                 return 1;
4413         }
4414
4415         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4416                           db_flags, &db);
4417         if (ret != 0) {
4418                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4419                 return ret;
4420         }
4421
4422         nodemap = get_nodemap(ctdb, false);
4423         if (nodemap == NULL) {
4424                 fprintf(stderr, "Failed to get nodemap\n");
4425                 return ENOMEM;
4426         }
4427
4428         ret = get_generation(mem_ctx, ctdb, &generation);
4429         if (ret != 0) {
4430                 fprintf(stderr, "Failed to get current generation\n");
4431                 return ret;
4432         }
4433
4434         count = list_of_active_nodes(nodemap, CTDB_UNKNOWN_PNN, mem_ctx,
4435                                      &pnn_list);
4436         if (count <= 0) {
4437                 return ENOMEM;
4438         }
4439
4440         ctdb_req_control_db_freeze(&request, db_id);
4441         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4442                                         ctdb->client, pnn_list, count,
4443                                         TIMEOUT(), &request, NULL, NULL);
4444         if (ret != 0) {
4445                 goto failed;
4446         }
4447
4448         wipedb.db_id = db_id;
4449         wipedb.tid = generation;
4450
4451         ctdb_req_control_db_transaction_start(&request, &wipedb);
4452         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4453                                         pnn_list, count, TIMEOUT(),
4454                                         &request, NULL, NULL);
4455         if (ret != 0) {
4456                 goto failed;
4457         }
4458
4459         ctdb_req_control_wipe_database(&request, &wipedb);
4460         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4461                                         pnn_list, count, TIMEOUT(),
4462                                         &request, NULL, NULL);
4463         if (ret != 0) {
4464                 goto failed;
4465         }
4466
4467         ctdb_req_control_db_set_healthy(&request, db_id);
4468         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4469                                         pnn_list, count, TIMEOUT(),
4470                                         &request, NULL, NULL);
4471         if (ret != 0) {
4472                 goto failed;
4473         }
4474
4475         ctdb_req_control_db_transaction_commit(&request, &wipedb);
4476         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4477                                         pnn_list, count, TIMEOUT(),
4478                                         &request, NULL, NULL);
4479         if (ret != 0) {
4480                 goto failed;
4481         }
4482
4483         ctdb_req_control_db_thaw(&request, db_id);
4484         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4485                                         ctdb->client, pnn_list, count,
4486                                         TIMEOUT(), &request, NULL, NULL);
4487         if (ret != 0) {
4488                 goto failed;
4489         }
4490
4491         printf("Database %s wiped\n", db_name);
4492         return 0;
4493
4494
4495 failed:
4496         ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
4497                               ctdb->pnn, TIMEOUT(), CTDB_RECOVERY_ACTIVE);
4498         return ret;
4499 }
4500
4501 static int control_recmaster(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4502                              int argc, const char **argv)
4503 {
4504         uint32_t recmaster;
4505         int ret;
4506
4507         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
4508                                       ctdb->cmd_pnn, TIMEOUT(), &recmaster);
4509         if (ret != 0) {
4510                 return ret;
4511         }
4512
4513         printf("%u\n", recmaster);
4514         return 0;
4515 }
4516
4517 static int control_event(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4518                          int argc, const char **argv)
4519 {
4520         char *t, *event_helper = NULL;
4521         char *eventd_socket = NULL;
4522         const char **new_argv;
4523         int i;
4524
4525         t = getenv("CTDB_EVENT_HELPER");
4526         if (t != NULL) {
4527                 event_helper = talloc_strdup(mem_ctx, t);
4528         } else {
4529                 event_helper = talloc_asprintf(mem_ctx, "%s/ctdb_event",
4530                                                CTDB_HELPER_BINDIR);
4531         }
4532
4533         if (event_helper == NULL) {
4534                 fprintf(stderr, "Unable to set event daemon helper\n");
4535                 return 1;
4536         }
4537
4538         t = getenv("CTDB_SOCKET");
4539         if (t != NULL) {
4540                 eventd_socket = talloc_asprintf(mem_ctx, "%s/eventd.sock",
4541                                                 dirname(t));
4542         } else {
4543                 eventd_socket = talloc_asprintf(mem_ctx, "%s/eventd.sock",
4544                                                 CTDB_RUNDIR);
4545         }
4546
4547         if (eventd_socket == NULL) {
4548                 fprintf(stderr, "Unable to set event daemon socket\n");
4549                 return 1;
4550         }
4551
4552         new_argv = talloc_array(mem_ctx, const char *, argc + 1);
4553         if (new_argv == NULL) {
4554                 fprintf(stderr, "Memory allocation error\n");
4555                 return 1;
4556         }
4557
4558         new_argv[0] = eventd_socket;
4559         for (i=0; i<argc; i++) {
4560                 new_argv[i+1] = argv[i];
4561         }
4562
4563         return run_helper(mem_ctx, "event daemon helper", event_helper,
4564                           argc+1, new_argv);
4565 }
4566
4567 static int control_scriptstatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4568                                 int argc, const char **argv)
4569 {
4570         const char *new_argv[3];
4571
4572         if (argc > 1) {
4573                 usage("scriptstatus");
4574         }
4575
4576         new_argv[0] = "status";
4577         new_argv[1] = (argc == 0) ? "monitor" : argv[0];
4578         new_argv[2] = NULL;
4579
4580         (void) control_event(mem_ctx, ctdb, 2, new_argv);
4581         return 0;
4582 }
4583
4584 static int control_natgw(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4585                          int argc, const char **argv)
4586 {
4587         char *t, *natgw_helper = NULL;
4588
4589         if (argc != 1) {
4590                 usage("natgw");
4591         }
4592
4593         t = getenv("CTDB_NATGW_HELPER");
4594         if (t != NULL) {
4595                 natgw_helper = talloc_strdup(mem_ctx, t);
4596         } else {
4597                 natgw_helper = talloc_asprintf(mem_ctx, "%s/ctdb_natgw",
4598                                                CTDB_HELPER_BINDIR);
4599         }
4600
4601         if (natgw_helper == NULL) {
4602                 fprintf(stderr, "Unable to set NAT gateway helper\n");
4603                 return 1;
4604         }
4605
4606         return run_helper(mem_ctx, "NAT gateway helper", natgw_helper,
4607                           argc, argv);
4608 }
4609
4610 /*
4611  * Find the PNN of the current node
4612  * discover the pnn by loading the nodes file and try to bind
4613  * to all addresses one at a time until the ip address is found.
4614  */
4615 static bool find_node_xpnn(TALLOC_CTX *mem_ctx, uint32_t *pnn)
4616 {
4617         struct ctdb_node_map *nodemap;
4618         int i;
4619
4620         nodemap = read_nodes_file(mem_ctx, CTDB_UNKNOWN_PNN);
4621         if (nodemap == NULL) {
4622                 return false;
4623         }
4624
4625         for (i=0; i<nodemap->num; i++) {
4626                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
4627                         continue;
4628                 }
4629                 if (ctdb_sys_have_ip(&nodemap->node[i].addr)) {
4630                         if (pnn != NULL) {
4631                                 *pnn = nodemap->node[i].pnn;
4632                         }
4633                         talloc_free(nodemap);
4634                         return true;
4635                 }
4636         }
4637
4638         fprintf(stderr, "Failed to detect PNN of the current node.\n");
4639         talloc_free(nodemap);
4640         return false;
4641 }
4642
4643 static int control_getreclock(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4644                               int argc, const char **argv)
4645 {
4646         const char *reclock;
4647         int ret;
4648
4649         if (argc != 0) {
4650                 usage("getreclock");
4651         }
4652
4653         ret = ctdb_ctrl_get_reclock_file(mem_ctx, ctdb->ev, ctdb->client,
4654                                          ctdb->cmd_pnn, TIMEOUT(), &reclock);
4655         if (ret != 0) {
4656                 return ret;
4657         }
4658
4659         if (reclock != NULL) {
4660                 printf("%s\n", reclock);
4661         }
4662
4663         return 0;
4664 }
4665
4666 static int control_setlmasterrole(TALLOC_CTX *mem_ctx,
4667                                   struct ctdb_context *ctdb,
4668                                   int argc, const char **argv)
4669 {
4670         uint32_t lmasterrole = 0;
4671         int ret;
4672
4673         if (argc != 1) {
4674                 usage("setlmasterrole");
4675         }
4676
4677         if (strcmp(argv[0], "on") == 0) {
4678                 lmasterrole = 1;
4679         } else if (strcmp(argv[0], "off") == 0) {
4680                 lmasterrole = 0;
4681         } else {
4682                 usage("setlmasterrole");
4683         }
4684
4685         ret = ctdb_ctrl_set_lmasterrole(mem_ctx, ctdb->ev, ctdb->client,
4686                                         ctdb->cmd_pnn, TIMEOUT(), lmasterrole);
4687         if (ret != 0) {
4688                 return ret;
4689         }
4690
4691         return 0;
4692 }
4693
4694 static int control_setrecmasterrole(TALLOC_CTX *mem_ctx,
4695                                     struct ctdb_context *ctdb,
4696                                     int argc, const char **argv)
4697 {
4698         uint32_t recmasterrole = 0;
4699         int ret;
4700
4701         if (argc != 1) {
4702                 usage("setrecmasterrole");
4703         }
4704
4705         if (strcmp(argv[0], "on") == 0) {
4706                 recmasterrole = 1;
4707         } else if (strcmp(argv[0], "off") == 0) {
4708                 recmasterrole = 0;
4709         } else {
4710                 usage("setrecmasterrole");
4711         }
4712
4713         ret = ctdb_ctrl_set_recmasterrole(mem_ctx, ctdb->ev, ctdb->client,
4714                                           ctdb->cmd_pnn, TIMEOUT(),
4715                                           recmasterrole);
4716         if (ret != 0) {
4717                 return ret;
4718         }
4719
4720         return 0;
4721 }
4722
4723 static int control_setdbreadonly(TALLOC_CTX *mem_ctx,
4724                                  struct ctdb_context *ctdb,
4725                                  int argc, const char **argv)
4726 {
4727         uint32_t db_id;
4728         uint8_t db_flags;
4729         int ret;
4730
4731         if (argc != 1) {
4732                 usage("setdbreadonly");
4733         }
4734
4735         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, NULL, &db_flags)) {
4736                 return 1;
4737         }
4738
4739         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
4740                 fprintf(stderr, "READONLY can be set only on volatile DB\n");
4741                 return 1;
4742         }
4743
4744         ret = ctdb_ctrl_set_db_readonly(mem_ctx, ctdb->ev, ctdb->client,
4745                                         ctdb->cmd_pnn, TIMEOUT(), db_id);
4746         if (ret != 0) {
4747                 return ret;
4748         }
4749
4750         return 0;
4751 }
4752
4753 static int control_setdbsticky(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4754                                int argc, const char **argv)
4755 {
4756         uint32_t db_id;
4757         uint8_t db_flags;
4758         int ret;
4759
4760         if (argc != 1) {
4761                 usage("setdbsticky");
4762         }
4763
4764         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, NULL, &db_flags)) {
4765                 return 1;
4766         }
4767
4768         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
4769                 fprintf(stderr, "STICKY can be set only on volatile DB\n");
4770                 return 1;
4771         }
4772
4773         ret = ctdb_ctrl_set_db_sticky(mem_ctx, ctdb->ev, ctdb->client,
4774                                       ctdb->cmd_pnn, TIMEOUT(), db_id);
4775         if (ret != 0) {
4776                 return ret;
4777         }
4778
4779         return 0;
4780 }
4781
4782 static int control_pfetch(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4783                           int argc, const char **argv)
4784 {
4785         const char *db_name;
4786         struct ctdb_db_context *db;
4787         struct ctdb_transaction_handle *h;
4788         uint8_t db_flags;
4789         TDB_DATA key, data;
4790         int ret;
4791
4792         if (argc < 2 || argc > 3) {
4793                 usage("pfetch");
4794         }
4795
4796         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
4797                 return 1;
4798         }
4799
4800         if (! (db_flags &
4801                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
4802                 fprintf(stderr, "Transactions not supported on DB %s\n",
4803                         db_name);
4804                 return 1;
4805         }
4806
4807         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4808                           db_flags, &db);
4809         if (ret != 0) {
4810                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4811                 return ret;
4812         }
4813
4814         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
4815         if (ret != 0) {
4816                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
4817                 return ret;
4818         }
4819
4820         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
4821                                      TIMEOUT(), db, true, &h);
4822         if (ret != 0) {
4823                 fprintf(stderr, "Failed to start transaction on db %s\n",
4824                         db_name);
4825                 return ret;
4826         }
4827
4828         ret = ctdb_transaction_fetch_record(h, key, mem_ctx, &data);
4829         if (ret != 0) {
4830                 fprintf(stderr, "Failed to read record for key %s\n",
4831                         argv[1]);
4832                 ctdb_transaction_cancel(h);
4833                 return ret;
4834         }
4835
4836         printf("%.*s\n", (int)data.dsize, data.dptr);
4837
4838         ctdb_transaction_cancel(h);
4839         return 0;
4840 }
4841
4842 static int control_pstore(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4843                           int argc, const char **argv)
4844 {
4845         const char *db_name;
4846         struct ctdb_db_context *db;
4847         struct ctdb_transaction_handle *h;
4848         uint8_t db_flags;
4849         TDB_DATA key, data;
4850         int ret;
4851
4852         if (argc != 3) {
4853                 usage("pstore");
4854         }
4855
4856         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
4857                 return 1;
4858         }
4859
4860         if (! (db_flags &
4861                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
4862                 fprintf(stderr, "Transactions not supported on DB %s\n",
4863                         db_name);
4864                 return 1;
4865         }
4866
4867         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4868                           db_flags, &db);
4869         if (ret != 0) {
4870                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4871                 return ret;
4872         }
4873
4874         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
4875         if (ret != 0) {
4876                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
4877                 return ret;
4878         }
4879
4880         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &data);
4881         if (ret != 0) {
4882                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
4883                 return ret;
4884         }
4885
4886         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
4887                                      TIMEOUT(), db, false, &h);
4888         if (ret != 0) {
4889                 fprintf(stderr, "Failed to start transaction on db %s\n",
4890                         db_name);
4891                 return ret;
4892         }
4893
4894         ret = ctdb_transaction_store_record(h, key, data);
4895         if (ret != 0) {
4896                 fprintf(stderr, "Failed to store record for key %s\n",
4897                         argv[1]);
4898                 ctdb_transaction_cancel(h);
4899                 return ret;
4900         }
4901
4902         ret = ctdb_transaction_commit(h);
4903         if (ret != 0) {
4904                 fprintf(stderr, "Failed to commit transaction on db %s\n",
4905                         db_name);
4906                 ctdb_transaction_cancel(h);
4907                 return ret;
4908         }
4909
4910         return 0;
4911 }
4912
4913 static int control_pdelete(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4914                            int argc, const char **argv)
4915 {
4916         const char *db_name;
4917         struct ctdb_db_context *db;
4918         struct ctdb_transaction_handle *h;
4919         uint8_t db_flags;
4920         TDB_DATA key;
4921         int ret;
4922
4923         if (argc != 2) {
4924                 usage("pdelete");
4925         }
4926
4927         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
4928                 return 1;
4929         }
4930
4931         if (! (db_flags &
4932                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
4933                 fprintf(stderr, "Transactions not supported on DB %s\n",
4934                         db_name);
4935                 return 1;
4936         }
4937
4938         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4939                           db_flags, &db);
4940         if (ret != 0) {
4941                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4942                 return ret;
4943         }
4944
4945         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
4946         if (ret != 0) {
4947                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
4948                 return ret;
4949         }
4950
4951         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
4952                                      TIMEOUT(), db, false, &h);
4953         if (ret != 0) {
4954                 fprintf(stderr, "Failed to start transaction on db %s\n",
4955                         db_name);
4956                 return ret;
4957         }
4958
4959         ret = ctdb_transaction_delete_record(h, key);
4960         if (ret != 0) {
4961                 fprintf(stderr, "Failed to delete record for key %s\n",
4962                         argv[1]);
4963                 ctdb_transaction_cancel(h);
4964                 return ret;
4965         }
4966
4967         ret = ctdb_transaction_commit(h);
4968         if (ret != 0) {
4969                 fprintf(stderr, "Failed to commit transaction on db %s\n",
4970                         db_name);
4971                 ctdb_transaction_cancel(h);
4972                 return ret;
4973         }
4974
4975         return 0;
4976 }
4977
4978 static int ptrans_parse_string(TALLOC_CTX *mem_ctx, const char **ptr, TDB_DATA *data)
4979 {
4980         const char *t;
4981         size_t n;
4982         int ret;
4983
4984         *data = tdb_null;
4985
4986         /* Skip whitespace */
4987         n = strspn(*ptr, " \t");
4988         t = *ptr + n;
4989
4990         if (t[0] == '"') {
4991                 /* Quoted ASCII string - no wide characters! */
4992                 t++;
4993                 n = strcspn(t, "\"");
4994                 if (t[n] == '"') {
4995                         if (n > 0) {
4996                                 ret = str_to_data(t, n, mem_ctx, data);
4997                                 if (ret != 0) {
4998                                         return ret;
4999                                 }
5000                         }
5001                         *ptr = t + n + 1;
5002                 } else {
5003                         fprintf(stderr, "Unmatched \" in input %s\n", *ptr);
5004                         return 1;
5005                 }
5006         } else {
5007                 fprintf(stderr, "Unsupported input format in %s\n", *ptr);
5008                 return 1;
5009         }
5010
5011         return 0;
5012 }
5013
5014 #define MAX_LINE_SIZE   1024
5015
5016 static bool ptrans_get_key_value(TALLOC_CTX *mem_ctx, FILE *file,
5017                                  TDB_DATA *key, TDB_DATA *value)
5018 {
5019         char line [MAX_LINE_SIZE]; /* FIXME: make this more flexible? */
5020         const char *ptr;
5021         int ret;
5022
5023         ptr = fgets(line, MAX_LINE_SIZE, file);
5024         if (ptr == NULL) {
5025                 return false;
5026         }
5027
5028         /* Get key */
5029         ret = ptrans_parse_string(mem_ctx, &ptr, key);
5030         if (ret != 0 || ptr == NULL || key->dptr == NULL) {
5031                 /* Line Ignored but not EOF */
5032                 *key = tdb_null;
5033                 return true;
5034         }
5035
5036         /* Get value */
5037         ret = ptrans_parse_string(mem_ctx, &ptr, value);
5038         if (ret != 0) {
5039                 /* Line Ignored but not EOF */
5040                 talloc_free(key->dptr);
5041                 *key = tdb_null;
5042                 return true;
5043         }
5044
5045         return true;
5046 }
5047
5048 static int control_ptrans(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5049                           int argc, const char **argv)
5050 {
5051         const char *db_name;
5052         struct ctdb_db_context *db;
5053         struct ctdb_transaction_handle *h;
5054         uint8_t db_flags;
5055         FILE *file;
5056         TDB_DATA key, value;
5057         int ret;
5058
5059         if (argc < 1 || argc > 2) {
5060                 usage("ptrans");
5061         }
5062
5063         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5064                 return 1;
5065         }
5066
5067         if (! (db_flags &
5068                (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED))) {
5069                 fprintf(stderr, "Transactions not supported on DB %s\n",
5070                         db_name);
5071                 return 1;
5072         }
5073
5074         if (argc == 2) {
5075                 file = fopen(argv[1], "r");
5076                 if (file == NULL) {
5077                         fprintf(stderr, "Failed to open file %s\n", argv[1]);
5078                         return 1;
5079                 }
5080         } else {
5081                 file = stdin;
5082         }
5083
5084         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5085                           db_flags, &db);
5086         if (ret != 0) {
5087                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5088                 goto done;
5089         }
5090
5091         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
5092                                      TIMEOUT(), db, false, &h);
5093         if (ret != 0) {
5094                 fprintf(stderr, "Failed to start transaction on db %s\n",
5095                         db_name);
5096                 goto done;
5097         }
5098
5099         while (ptrans_get_key_value(mem_ctx, file, &key, &value)) {
5100                 if (key.dsize != 0) {
5101                         ret = ctdb_transaction_store_record(h, key, value);
5102                         if (ret != 0) {
5103                                 fprintf(stderr, "Failed to store record\n");
5104                                 ctdb_transaction_cancel(h);
5105                                 goto done;
5106                         }
5107                         talloc_free(key.dptr);
5108                         talloc_free(value.dptr);
5109                 }
5110         }
5111
5112         ret = ctdb_transaction_commit(h);
5113         if (ret != 0) {
5114                 fprintf(stderr, "Failed to commit transaction on db %s\n",
5115                         db_name);
5116                 ctdb_transaction_cancel(h);
5117         }
5118
5119 done:
5120         if (file != stdin) {
5121                 fclose(file);
5122         }
5123         return ret;
5124 }
5125
5126 static int control_tfetch(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5127                           int argc, const char **argv)
5128 {
5129         struct tdb_context *tdb;
5130         TDB_DATA key, data;
5131         struct ctdb_ltdb_header header;
5132         int ret;
5133
5134         if (argc < 2 || argc > 3) {
5135                 usage("tfetch");
5136         }
5137
5138         tdb = tdb_open(argv[0], 0, 0, O_RDWR, 0);
5139         if (tdb == NULL) {
5140                 fprintf(stderr, "Failed to open TDB file %s\n", argv[0]);
5141                 return 1;
5142         }
5143
5144         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5145         if (ret != 0) {
5146                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5147                 tdb_close(tdb);
5148                 return ret;
5149         }
5150
5151         data = tdb_fetch(tdb, key);
5152         if (data.dptr == NULL) {
5153                 fprintf(stderr, "No record for key %s\n", argv[1]);
5154                 tdb_close(tdb);
5155                 return 1;
5156         }
5157
5158         if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
5159                 fprintf(stderr, "Invalid record for key %s\n", argv[1]);
5160                 tdb_close(tdb);
5161                 return 1;
5162         }
5163
5164         tdb_close(tdb);
5165
5166         if (argc == 3) {
5167                 int fd;
5168                 ssize_t nwritten;
5169
5170                 fd = open(argv[2], O_WRONLY|O_CREAT|O_TRUNC, 0600);
5171                 if (fd == -1) {
5172                         fprintf(stderr, "Failed to open output file %s\n",
5173                                 argv[2]);
5174                         goto fail;
5175                 }
5176
5177                 nwritten = sys_write(fd, data.dptr, data.dsize);
5178                 if (nwritten != data.dsize) {
5179                         fprintf(stderr, "Failed to write record to file\n");
5180                         close(fd);
5181                         goto fail;
5182                 }
5183
5184                 close(fd);
5185         }
5186
5187 fail:
5188         ret = ctdb_ltdb_header_extract(&data, &header);
5189         if (ret != 0) {
5190                 fprintf(stderr, "Failed to parse header from data\n");
5191                 return 1;
5192         }
5193
5194         dump_ltdb_header(&header);
5195         dump_tdb_data("data", data);
5196
5197         return 0;
5198 }
5199
5200 static int control_tstore(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5201                           int argc, const char **argv)
5202 {
5203         struct tdb_context *tdb;
5204         TDB_DATA key, data[2], value;
5205         struct ctdb_ltdb_header header;
5206         uint8_t header_buf[sizeof(struct ctdb_ltdb_header)];
5207         size_t np;
5208         int ret;
5209
5210         if (argc < 3 || argc > 5) {
5211                 usage("tstore");
5212         }
5213
5214         tdb = tdb_open(argv[0], 0, 0, O_RDWR, 0);
5215         if (tdb == NULL) {
5216                 fprintf(stderr, "Failed to open TDB file %s\n", argv[0]);
5217                 return 1;
5218         }
5219
5220         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5221         if (ret != 0) {
5222                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5223                 tdb_close(tdb);
5224                 return ret;
5225         }
5226
5227         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &value);
5228         if (ret != 0) {
5229                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
5230                 tdb_close(tdb);
5231                 return ret;
5232         }
5233
5234         ZERO_STRUCT(header);
5235
5236         if (argc > 3) {
5237                 header.rsn = (uint64_t)strtoull(argv[3], NULL, 0);
5238         }
5239         if (argc > 4) {
5240                 header.dmaster = (uint32_t)atol(argv[4]);
5241         }
5242         if (argc > 5) {
5243                 header.flags = (uint32_t)atol(argv[5]);
5244         }
5245
5246         ctdb_ltdb_header_push(&header, header_buf, &np);
5247
5248         data[0].dsize = np;
5249         data[0].dptr = header_buf;
5250
5251         data[1].dsize = value.dsize;
5252         data[1].dptr = value.dptr;
5253
5254         ret = tdb_storev(tdb, key, data, 2, TDB_REPLACE);
5255         if (ret != 0) {
5256                 fprintf(stderr, "Failed to write record %s to file %s\n",
5257                         argv[1], argv[0]);
5258         }
5259
5260         tdb_close(tdb);
5261
5262         return ret;
5263 }
5264
5265 static int control_readkey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5266                            int argc, const char **argv)
5267 {
5268         const char *db_name;
5269         struct ctdb_db_context *db;
5270         struct ctdb_record_handle *h;
5271         uint8_t db_flags;
5272         TDB_DATA key, data;
5273         bool readonly = false;
5274         int ret;
5275
5276         if (argc < 2 || argc > 3) {
5277                 usage("readkey");
5278         }
5279
5280         if (argc == 3) {
5281                 if (strcmp(argv[2], "readonly") == 0) {
5282                         readonly = true;
5283                 } else {
5284                         usage("readkey");
5285                 }
5286         }
5287
5288         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5289                 return 1;
5290         }
5291
5292         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
5293                 fprintf(stderr, "DB %s is not a volatile database\n",
5294                         db_name);
5295                 return 1;
5296         }
5297
5298         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5299                           db_flags, &db);
5300         if (ret != 0) {
5301                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5302                 return ret;
5303         }
5304
5305         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5306         if (ret != 0) {
5307                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5308                 return ret;
5309         }
5310
5311         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5312                               db, key, readonly, &h, NULL, &data);
5313         if (ret != 0) {
5314                 fprintf(stderr, "Failed to read record for key %s\n",
5315                         argv[1]);
5316         } else {
5317                 printf("Data: size:%zu ptr:[%.*s]\n", data.dsize,
5318                        (int)data.dsize, data.dptr);
5319         }
5320
5321         talloc_free(h);
5322         return ret;
5323 }
5324
5325 static int control_writekey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5326                             int argc, const char **argv)
5327 {
5328         const char *db_name;
5329         struct ctdb_db_context *db;
5330         struct ctdb_record_handle *h;
5331         uint8_t db_flags;
5332         TDB_DATA key, data;
5333         int ret;
5334
5335         if (argc != 3) {
5336                 usage("writekey");
5337         }
5338
5339         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5340                 return 1;
5341         }
5342
5343         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
5344                 fprintf(stderr, "DB %s is not a volatile database\n",
5345                         db_name);
5346                 return 1;
5347         }
5348
5349         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5350                           db_flags, &db);
5351         if (ret != 0) {
5352                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5353                 return ret;
5354         }
5355
5356         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5357         if (ret != 0) {
5358                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5359                 return ret;
5360         }
5361
5362         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &data);
5363         if (ret != 0) {
5364                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
5365                 return ret;
5366         }
5367
5368         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5369                               db, key, false, &h, NULL, NULL);
5370         if (ret != 0) {
5371                 fprintf(stderr, "Failed to lock record for key %s\n", argv[0]);
5372                 return ret;
5373         }
5374
5375         ret = ctdb_store_record(h, data);
5376         if (ret != 0) {
5377                 fprintf(stderr, "Failed to store record for key %s\n",
5378                         argv[1]);
5379         }
5380
5381         talloc_free(h);
5382         return ret;
5383 }
5384
5385 static int control_deletekey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5386                              int argc, const char **argv)
5387 {
5388         const char *db_name;
5389         struct ctdb_db_context *db;
5390         struct ctdb_record_handle *h;
5391         uint8_t db_flags;
5392         TDB_DATA key, data;
5393         int ret;
5394
5395         if (argc != 2) {
5396                 usage("deletekey");
5397         }
5398
5399         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5400                 return 1;
5401         }
5402
5403         if (db_flags & (CTDB_DB_FLAGS_PERSISTENT | CTDB_DB_FLAGS_REPLICATED)) {
5404                 fprintf(stderr, "DB %s is not a volatile database\n",
5405                         db_name);
5406                 return 1;
5407         }
5408
5409         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5410                           db_flags, &db);
5411         if (ret != 0) {
5412                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5413                 return ret;
5414         }
5415
5416         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5417         if (ret != 0) {
5418                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5419                 return ret;
5420         }
5421
5422         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5423                               db, key, false, &h, NULL, &data);
5424         if (ret != 0) {
5425                 fprintf(stderr, "Failed to fetch record for key %s\n",
5426                         argv[1]);
5427                 return ret;
5428         }
5429
5430         ret = ctdb_delete_record(h);
5431         if (ret != 0) {
5432                 fprintf(stderr, "Failed to delete record for key %s\n",
5433                         argv[1]);
5434         }
5435
5436         talloc_free(h);
5437         return ret;
5438 }
5439
5440 static int control_checktcpport(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5441                                 int argc, const char **argv)
5442 {
5443         struct sockaddr_in sin;
5444         unsigned int port;
5445         int s, v;
5446         int ret;
5447
5448         if (argc != 1) {
5449                 usage("chktcpport");
5450         }
5451
5452         port = atoi(argv[0]);
5453
5454         s = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
5455         if (s == -1) {
5456                 fprintf(stderr, "Failed to open local socket\n");
5457                 return errno;
5458         }
5459
5460         v = fcntl(s, F_GETFL, 0);
5461         if (v == -1 || fcntl(s, F_SETFL, v | O_NONBLOCK)) {
5462                 fprintf(stderr, "Unable to set socket non-blocking\n");
5463                 close(s);
5464                 return errno;
5465         }
5466
5467         bzero(&sin, sizeof(sin));
5468         sin.sin_family = AF_INET;
5469         sin.sin_port = htons(port);
5470         ret = bind(s, (struct sockaddr *)&sin, sizeof(sin));
5471         close(s);
5472         if (ret == -1) {
5473                 fprintf(stderr, "Failed to bind to TCP port %u\n", port);
5474                 return errno;
5475         }
5476
5477         return 0;
5478 }
5479
5480 static int control_getdbseqnum(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5481                                int argc, const char **argv)
5482 {
5483         uint32_t db_id;
5484         const char *db_name;
5485         uint64_t seqnum;
5486         int ret;
5487
5488         if (argc != 1) {
5489                 usage("getdbseqnum");
5490         }
5491
5492         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, NULL)) {
5493                 return 1;
5494         }
5495
5496         ret = ctdb_ctrl_get_db_seqnum(mem_ctx, ctdb->ev, ctdb->client,
5497                                       ctdb->cmd_pnn, TIMEOUT(), db_id,
5498                                       &seqnum);
5499         if (ret != 0) {
5500                 fprintf(stderr, "Failed to get sequence number for DB %s\n",
5501                         db_name);
5502                 return ret;
5503         }
5504
5505         printf("0x%"PRIx64"\n", seqnum);
5506         return 0;
5507 }
5508
5509 static int control_nodestatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5510                               int argc, const char **argv)
5511 {
5512         const char *nodestring = NULL;
5513         struct ctdb_node_map *nodemap;
5514         int ret, i;
5515         bool print_hdr = false;
5516
5517         if (argc > 1) {
5518                 usage("nodestatus");
5519         }
5520
5521         if (argc == 1) {
5522                 nodestring = argv[0];
5523                 if (strcmp(nodestring, "all") == 0) {
5524                         print_hdr = true;
5525                 }
5526         }
5527
5528         if (! parse_nodestring(mem_ctx, ctdb, nodestring, &nodemap)) {
5529                 return 1;
5530         }
5531
5532         if (options.machinereadable) {
5533                 print_nodemap_machine(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn);
5534         } else {
5535                 print_nodemap(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn, print_hdr);
5536         }
5537
5538         ret = 0;
5539         for (i=0; i<nodemap->num; i++) {
5540                 ret |= nodemap->node[i].flags;
5541         }
5542
5543         return ret;
5544 }
5545
5546 const struct {
5547         const char *name;
5548         uint32_t offset;
5549 } db_stats_fields[] = {
5550 #define DBSTATISTICS_FIELD(n) { #n, offsetof(struct ctdb_db_statistics, n) }
5551         DBSTATISTICS_FIELD(db_ro_delegations),
5552         DBSTATISTICS_FIELD(db_ro_revokes),
5553         DBSTATISTICS_FIELD(locks.num_calls),
5554         DBSTATISTICS_FIELD(locks.num_current),
5555         DBSTATISTICS_FIELD(locks.num_pending),
5556         DBSTATISTICS_FIELD(locks.num_failed),
5557         DBSTATISTICS_FIELD(db_ro_delegations),
5558 };
5559
5560 static void print_dbstatistics(const char *db_name,
5561                                struct ctdb_db_statistics *s)
5562 {
5563         int i;
5564         const char *prefix = NULL;
5565         int preflen = 0;
5566
5567         printf("DB Statistics %s\n", db_name);
5568
5569         for (i=0; i<ARRAY_SIZE(db_stats_fields); i++) {
5570                 if (strchr(db_stats_fields[i].name, '.') != NULL) {
5571                         preflen = strcspn(db_stats_fields[i].name, ".") + 1;
5572                         if (! prefix ||
5573                             strncmp(prefix, db_stats_fields[i].name, preflen) != 0) {
5574                                 prefix = db_stats_fields[i].name;
5575                                 printf(" %*.*s\n", preflen-1, preflen-1,
5576                                        db_stats_fields[i].name);
5577                         }
5578                 } else {
5579                         preflen = 0;
5580                 }
5581                 printf(" %*s%-22s%*s%10u\n", preflen ? 4 : 0, "",
5582                        db_stats_fields[i].name+preflen, preflen ? 0 : 4, "",
5583                        *(uint32_t *)(db_stats_fields[i].offset+(uint8_t *)s));
5584         }
5585
5586         printf(" hop_count_buckets:");
5587         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
5588                 printf(" %d", s->hop_count_bucket[i]);
5589         }
5590         printf("\n");
5591
5592         printf(" lock_buckets:");
5593         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
5594                 printf(" %d", s->locks.buckets[i]);
5595         }
5596         printf("\n");
5597
5598         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
5599                "locks_latency      MIN/AVG/MAX",
5600                s->locks.latency.min, LATENCY_AVG(s->locks.latency),
5601                s->locks.latency.max, s->locks.latency.num);
5602
5603         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
5604                "vacuum_latency     MIN/AVG/MAX",
5605                s->vacuum.latency.min, LATENCY_AVG(s->vacuum.latency),
5606                s->vacuum.latency.max, s->vacuum.latency.num);
5607
5608         printf(" Num Hot Keys:     %d\n", s->num_hot_keys);
5609         for (i=0; i<s->num_hot_keys; i++) {
5610                 int j;
5611                 printf("     Count:%d Key:", s->hot_keys[i].count);
5612                 for (j=0; j<s->hot_keys[i].key.dsize; j++) {
5613                         printf("%02x", s->hot_keys[i].key.dptr[j] & 0xff);
5614                 }
5615                 printf("\n");
5616         }
5617 }
5618
5619 static int control_dbstatistics(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5620                                 int argc, const char **argv)
5621 {
5622         uint32_t db_id;
5623         const char *db_name;
5624         struct ctdb_db_statistics *dbstats;
5625         int ret;
5626
5627         if (argc != 1) {
5628                 usage("dbstatistics");
5629         }
5630
5631         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, NULL)) {
5632                 return 1;
5633         }
5634
5635         ret = ctdb_ctrl_get_db_statistics(mem_ctx, ctdb->ev, ctdb->client,
5636                                           ctdb->cmd_pnn, TIMEOUT(), db_id,
5637                                           &dbstats);
5638         if (ret != 0) {
5639                 fprintf(stderr, "Failed to get statistics for DB %s\n",
5640                         db_name);
5641                 return ret;
5642         }
5643
5644         print_dbstatistics(db_name, dbstats);
5645         return 0;
5646 }
5647
5648 struct disable_takeover_runs_state {
5649         uint32_t *pnn_list;
5650         int node_count;
5651         bool *reply;
5652         int status;
5653         bool done;
5654 };
5655
5656 static void disable_takeover_run_handler(uint64_t srvid, TDB_DATA data,
5657                                          void *private_data)
5658 {
5659         struct disable_takeover_runs_state *state =
5660                 (struct disable_takeover_runs_state *)private_data;
5661         int ret, i;
5662
5663         if (data.dsize != sizeof(int)) {
5664                 /* Ignore packet */
5665                 return;
5666         }
5667
5668         /* ret will be a PNN (i.e. >=0) on success, or negative on error */
5669         ret = *(int *)data.dptr;
5670         if (ret < 0) {
5671                 state->status = ret;
5672                 state->done = true;
5673                 return;
5674         }
5675         for (i=0; i<state->node_count; i++) {
5676                 if (state->pnn_list[i] == ret) {
5677                         state->reply[i] = true;
5678                         break;
5679                 }
5680         }
5681
5682         state->done = true;
5683         for (i=0; i<state->node_count; i++) {
5684                 if (! state->reply[i]) {
5685                         state->done = false;
5686                         break;
5687                 }
5688         }
5689 }
5690
5691 static int disable_takeover_runs(TALLOC_CTX *mem_ctx,
5692                                  struct ctdb_context *ctdb, uint32_t timeout,
5693                                  uint32_t *pnn_list, int count)
5694 {
5695         struct ctdb_disable_message disable = { 0 };
5696         struct disable_takeover_runs_state state;
5697         int ret, i;
5698
5699         disable.pnn = ctdb->pnn;
5700         disable.srvid = next_srvid(ctdb);
5701         disable.timeout = timeout;
5702
5703         state.pnn_list = pnn_list;
5704         state.node_count = count;
5705         state.done = false;
5706         state.status = 0;
5707         state.reply = talloc_zero_array(mem_ctx, bool, count);
5708         if (state.reply == NULL) {
5709                 return ENOMEM;
5710         }
5711
5712         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
5713                                               disable.srvid,
5714                                               disable_takeover_run_handler,
5715                                               &state);
5716         if (ret != 0) {
5717                 return ret;
5718         }
5719
5720         for (i=0; i<count; i++) {
5721                 ret = ctdb_message_disable_takeover_runs(mem_ctx, ctdb->ev,
5722                                                          ctdb->client,
5723                                                          pnn_list[i],
5724                                                          &disable);
5725                 if (ret != 0) {
5726                         goto fail;
5727                 }
5728         }
5729
5730         ret = ctdb_client_wait_timeout(ctdb->ev, &state.done, TIMEOUT());
5731         if (ret == ETIME) {
5732                 fprintf(stderr, "Timed out waiting to disable takeover runs\n");
5733         } else {
5734                 ret = (state.status >= 0 ? 0 : 1);
5735         }
5736
5737 fail:
5738         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
5739                                            disable.srvid, &state);
5740         return ret;
5741 }
5742
5743 static int control_reloadips(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5744                              int argc, const char **argv)
5745 {
5746         const char *nodestring = NULL;
5747         struct ctdb_node_map *nodemap, *nodemap2;
5748         struct ctdb_req_control request;
5749         uint32_t *pnn_list, *pnn_list2;
5750         int ret, count, count2;
5751
5752         if (argc > 1) {
5753                 usage("reloadips");
5754         }
5755
5756         if (argc == 1) {
5757                 nodestring = argv[0];
5758         }
5759
5760         nodemap = get_nodemap(ctdb, false);
5761         if (nodemap == NULL) {
5762                 return 1;
5763         }
5764
5765         if (! parse_nodestring(mem_ctx, ctdb, nodestring, &nodemap2)) {
5766                 return 1;
5767         }
5768
5769         count = list_of_connected_nodes(nodemap, CTDB_UNKNOWN_PNN,
5770                                         mem_ctx, &pnn_list);
5771         if (count <= 0) {
5772                 fprintf(stderr, "Memory allocation error\n");
5773                 return 1;
5774         }
5775
5776         count2 = list_of_active_nodes(nodemap2, CTDB_UNKNOWN_PNN,
5777                                       mem_ctx, &pnn_list2);
5778         if (count2 <= 0) {
5779                 fprintf(stderr, "Memory allocation error\n");
5780                 return 1;
5781         }
5782
5783         /* Disable takeover runs on all connected nodes.  A reply
5784          * indicating success is needed from each node so all nodes
5785          * will need to be active.
5786          *
5787          * A check could be added to not allow reloading of IPs when
5788          * there are disconnected nodes.  However, this should
5789          * probably be left up to the administrator.
5790          */
5791         ret = disable_takeover_runs(mem_ctx, ctdb, 2*options.timelimit,
5792                                     pnn_list, count);
5793         if (ret != 0) {
5794                 fprintf(stderr, "Failed to disable takeover runs\n");
5795                 return ret;
5796         }
5797
5798         /* Now tell all the desired nodes to reload their public IPs.
5799          * Keep trying this until it succeeds.  This assumes all
5800          * failures are transient, which might not be true...
5801          */
5802         ctdb_req_control_reload_public_ips(&request);
5803         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
5804                                         pnn_list2, count2, TIMEOUT(),
5805                                         &request, NULL, NULL);
5806         if (ret != 0) {
5807                 fprintf(stderr, "Failed to reload IPs on some nodes.\n");
5808         }
5809
5810         /* It isn't strictly necessary to wait until takeover runs are
5811          * re-enabled but doing so can't hurt.
5812          */
5813         ret = disable_takeover_runs(mem_ctx, ctdb, 0, pnn_list, count);
5814         if (ret != 0) {
5815                 fprintf(stderr, "Failed to enable takeover runs\n");
5816                 return ret;
5817         }
5818
5819         return ipreallocate(mem_ctx, ctdb);
5820 }
5821
5822 static int control_ipiface(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5823                            int argc, const char **argv)
5824 {
5825         ctdb_sock_addr addr;
5826         char *iface;
5827
5828         if (argc != 1) {
5829                 usage("ipiface");
5830         }
5831
5832         if (! parse_ip(argv[0], NULL, 0, &addr)) {
5833                 fprintf(stderr, "Failed to Parse IP %s\n", argv[0]);
5834                 return 1;
5835         }
5836
5837         iface = ctdb_sys_find_ifname(&addr);
5838         if (iface == NULL) {
5839                 fprintf(stderr, "Failed to find interface for IP %s\n",
5840                         argv[0]);
5841                 return 1;
5842         }
5843         free(iface);
5844
5845         return 0;
5846 }
5847
5848
5849 static const struct ctdb_cmd {
5850         const char *name;
5851         int (*fn)(TALLOC_CTX *, struct ctdb_context *, int, const char **);
5852         bool without_daemon; /* can be run without daemon running ? */
5853         bool remote; /* can be run on remote nodes */
5854         const char *msg;
5855         const char *args;
5856 } ctdb_commands[] = {
5857         { "version", control_version, true, false,
5858                 "show version of ctdb", NULL },
5859         { "status", control_status, false, true,
5860                 "show node status", NULL },
5861         { "uptime", control_uptime, false, true,
5862                 "show node uptime", NULL },
5863         { "ping", control_ping, false, true,
5864                 "ping all nodes", NULL },
5865         { "runstate", control_runstate, false, true,
5866                 "get/check runstate of a node",
5867                 "[setup|first_recovery|startup|running]" },
5868         { "getvar", control_getvar, false, true,
5869                 "get a tunable variable", "<name>" },
5870         { "setvar", control_setvar, false, true,
5871                 "set a tunable variable", "<name> <value>" },
5872         { "listvars", control_listvars, false, true,
5873                 "list tunable variables", NULL },
5874         { "statistics", control_statistics, false, true,
5875                 "show ctdb statistics", NULL },
5876         { "statisticsreset", control_statistics_reset, false, true,
5877                 "reset ctdb statistics", NULL },
5878         { "stats", control_stats, false, true,
5879                 "show rolling statistics", "[count]" },
5880         { "ip", control_ip, false, true,
5881                 "show public ips", "[all]" },
5882         { "ipinfo", control_ipinfo, false, true,
5883                 "show public ip details", "<ip>" },
5884         { "ifaces", control_ifaces, false, true,
5885                 "show interfaces", NULL },
5886         { "setifacelink", control_setifacelink, false, true,
5887                 "set interface link status", "<iface> up|down" },
5888         { "process-exists", control_process_exists, false, true,
5889                 "check if a process exists on a node",  "<pid>" },
5890         { "getdbmap", control_getdbmap, false, true,
5891                 "show attached databases", NULL },
5892         { "getdbstatus", control_getdbstatus, false, true,
5893                 "show database status", "<dbname|dbid>" },
5894         { "catdb", control_catdb, false, false,
5895                 "dump cluster-wide ctdb database", "<dbname|dbid>" },
5896         { "cattdb", control_cattdb, false, false,
5897                 "dump local ctdb database", "<dbname|dbid>" },
5898         { "getcapabilities", control_getcapabilities, false, true,
5899                 "show node capabilities", NULL },
5900         { "pnn", control_pnn, false, false,
5901                 "show the pnn of the currnet node", NULL },
5902         { "lvs", control_lvs, false, false,
5903                 "show lvs configuration", "master|list|status" },
5904         { "setdebug", control_setdebug, false, true,
5905                 "set debug level", "ERROR|WARNING|NOTICE|INFO|DEBUG" },
5906         { "getdebug", control_getdebug, false, true,
5907                 "get debug level", NULL },
5908         { "attach", control_attach, false, false,
5909                 "attach a database", "<dbname> [persistent|replicated]" },
5910         { "detach", control_detach, false, false,
5911                 "detach database(s)", "<dbname|dbid> ..." },
5912         { "dumpmemory", control_dumpmemory, false, true,
5913                 "dump ctdbd memory map", NULL },
5914         { "rddumpmemory", control_rddumpmemory, false, true,
5915                 "dump recoverd memory map", NULL },
5916         { "getpid", control_getpid, false, true,
5917                 "get ctdbd process ID", NULL },
5918         { "disable", control_disable, false, true,
5919                 "disable a node", NULL },
5920         { "enable", control_enable, false, true,
5921                 "enable a node", NULL },
5922         { "stop", control_stop, false, true,
5923                 "stop a node", NULL },
5924         { "continue", control_continue, false, true,
5925                 "continue a stopped node", NULL },
5926         { "ban", control_ban, false, true,
5927                 "ban a node", "<bantime>"},
5928         { "unban", control_unban, false, true,
5929                 "unban a node", NULL },
5930         { "shutdown", control_shutdown, false, true,
5931                 "shutdown ctdb daemon", NULL },
5932         { "recover", control_recover, false, true,
5933                 "force recovery", NULL },
5934         { "sync", control_ipreallocate, false, true,
5935                 "run ip reallocation (deprecated)", NULL },
5936         { "ipreallocate", control_ipreallocate, false, true,
5937                 "run ip reallocation", NULL },
5938         { "isnotrecmaster", control_isnotrecmaster, false, false,
5939                 "check if local node is the recmaster", NULL },
5940         { "gratarp", control_gratarp, false, true,
5941                 "send a gratuitous arp", "<ip> <interface>" },
5942         { "tickle", control_tickle, true, false,
5943                 "send a tcp tickle ack", "<srcip:port> <dstip:port>" },
5944         { "gettickles", control_gettickles, false, true,
5945                 "get the list of tickles", "<ip> [<port>]" },
5946         { "addtickle", control_addtickle, false, true,
5947                 "add a tickle", "<ip>:<port> <ip>:<port>" },
5948         { "deltickle", control_deltickle, false, true,
5949                 "delete a tickle", "<ip>:<port> <ip>:<port>" },
5950         { "listnodes", control_listnodes, true, true,
5951                 "list nodes in the cluster", NULL },
5952         { "reloadnodes", control_reloadnodes, false, false,
5953                 "reload the nodes file all nodes", NULL },
5954         { "moveip", control_moveip, false, false,
5955                 "move an ip address to another node", "<ip> <node>" },
5956         { "addip", control_addip, false, true,
5957                 "add an ip address to a node", "<ip/mask> <iface>" },
5958         { "delip", control_delip, false, true,
5959                 "delete an ip address from a node", "<ip>" },
5960         { "backupdb", control_backupdb, false, false,
5961                 "backup a database into a file", "<dbname|dbid> <file>" },
5962         { "restoredb", control_restoredb, false, false,
5963                 "restore a database from a file", "<file> [dbname]" },
5964         { "dumpdbbackup", control_dumpdbbackup, true, false,
5965                 "dump database from a backup file", "<file>" },
5966         { "wipedb", control_wipedb, false, false,
5967                 "wipe the contents of a database.", "<dbname|dbid>"},
5968         { "recmaster", control_recmaster, false, true,
5969                 "show the pnn for the recovery master", NULL },
5970         { "event", control_event, true, false,
5971                 "event and event script commands", NULL },
5972         { "scriptstatus", control_scriptstatus, true, false,
5973                 "show event script status",
5974                 "[init|setup|startup|monitor|takeip|releaseip|ipreallocated]" },
5975         { "natgw", control_natgw, false, false,
5976                 "show natgw configuration", "master|list|status" },
5977         { "getreclock", control_getreclock, false, true,
5978                 "get recovery lock file", NULL },
5979         { "setlmasterrole", control_setlmasterrole, false, true,
5980                 "set LMASTER role", "on|off" },
5981         { "setrecmasterrole", control_setrecmasterrole, false, true,
5982                 "set RECMASTER role", "on|off"},
5983         { "setdbreadonly", control_setdbreadonly, false, true,
5984                 "enable readonly records", "<dbname|dbid>" },
5985         { "setdbsticky", control_setdbsticky, false, true,
5986                 "enable sticky records", "<dbname|dbid>"},
5987         { "pfetch", control_pfetch, false, false,
5988                 "fetch record from persistent database", "<dbname|dbid> <key> [<file>]" },
5989         { "pstore", control_pstore, false, false,
5990                 "write record to persistent database", "<dbname|dbid> <key> <value>" },
5991         { "pdelete", control_pdelete, false, false,
5992                 "delete record from persistent database", "<dbname|dbid> <key>" },
5993         { "ptrans", control_ptrans, false, false,
5994                 "update a persistent database (from file or stdin)", "<dbname|dbid> [<file>]" },
5995         { "tfetch", control_tfetch, false, true,
5996                 "fetch a record", "<tdb-file> <key> [<file>]" },
5997         { "tstore", control_tstore, false, true,
5998                 "store a record", "<tdb-file> <key> <data> [<rsn> <dmaster> <flags>]" },
5999         { "readkey", control_readkey, false, false,
6000                 "read value of a database key", "<dbname|dbid> <key> [readonly]" },
6001         { "writekey", control_writekey, false, false,
6002                 "write value for a database key", "<dbname|dbid> <key> <value>" },
6003         { "deletekey", control_deletekey, false, false,
6004                 "delete a database key", "<dbname|dbid> <key>" },
6005         { "checktcpport", control_checktcpport, true, false,
6006                 "check if a service is bound to a specific tcp port or not", "<port>" },
6007         { "getdbseqnum", control_getdbseqnum, false, false,
6008                 "get database sequence number", "<dbname|dbid>" },
6009         { "nodestatus", control_nodestatus, false, true,
6010                 "show and return node status", "[all|<pnn-list>]" },
6011         { "dbstatistics", control_dbstatistics, false, true,
6012                 "show database statistics", "<dbname|dbid>" },
6013         { "reloadips", control_reloadips, false, false,
6014                 "reload the public addresses file", "[all|<pnn-list>]" },
6015         { "ipiface", control_ipiface, true, false,
6016                 "Find the interface an ip address is hosted on", "<ip>" },
6017 };
6018
6019 static const struct ctdb_cmd *match_command(const char *command)
6020 {
6021         const struct ctdb_cmd *cmd;
6022         int i;
6023
6024         for (i=0; i<ARRAY_SIZE(ctdb_commands); i++) {
6025                 cmd = &ctdb_commands[i];
6026                 if (strlen(command) == strlen(cmd->name) &&
6027                     strncmp(command, cmd->name, strlen(command)) == 0) {
6028                         return cmd;
6029                 }
6030         }
6031
6032         return NULL;
6033 }
6034
6035
6036 /**
6037  * Show usage message
6038  */
6039 static void usage_full(void)
6040 {
6041         int i;
6042
6043         poptPrintHelp(pc, stdout, 0);
6044         printf("\nCommands:\n");
6045         for (i=0; i<ARRAY_SIZE(ctdb_commands); i++) {
6046                 printf("  %-15s %-27s  %s\n",
6047                        ctdb_commands[i].name,
6048                        ctdb_commands[i].args ? ctdb_commands[i].args : "",
6049                        ctdb_commands[i].msg);
6050         }
6051 }
6052
6053 static void usage(const char *command)
6054 {
6055         const struct ctdb_cmd *cmd;
6056
6057         if (command == NULL) {
6058                 usage_full();
6059                 exit(1);
6060         }
6061
6062         cmd = match_command(command);
6063         if (cmd == NULL) {
6064                 usage_full();
6065         } else {
6066                 poptPrintUsage(pc, stdout, 0);
6067                 printf("\nCommands:\n");
6068                 printf("  %-15s %-27s  %s\n",
6069                        cmd->name, cmd->args ? cmd->args : "", cmd->msg);
6070         }
6071
6072         exit(1);
6073 }
6074
6075 struct poptOption cmdline_options[] = {
6076         POPT_AUTOHELP
6077         { "socket", 's', POPT_ARG_STRING, &options.socket, 0,
6078                 "CTDB socket path", "filename" },
6079         { "debug", 'd', POPT_ARG_STRING, &options.debuglevelstr, 0,
6080                 "debug level"},
6081         { "timelimit", 't', POPT_ARG_INT, &options.timelimit, 0,
6082                 "timelimit (in seconds)" },
6083         { "node", 'n', POPT_ARG_INT, &options.pnn, 0,
6084                 "node specification - integer" },
6085         { NULL, 'Y', POPT_ARG_NONE, &options.machinereadable, 0,
6086                 "enable machine readable output", NULL },
6087         { "separator", 'x', POPT_ARG_STRING, &options.sep, 0,
6088                 "specify separator for machine readable output", "CHAR" },
6089         { NULL, 'X', POPT_ARG_NONE, &options.machineparsable, 0,
6090                 "enable machine parsable output with separator |", NULL },
6091         { "verbose", 'v', POPT_ARG_NONE, &options.verbose, 0,
6092                 "enable verbose output", NULL },
6093         { "maxruntime", 'T', POPT_ARG_INT, &options.maxruntime, 0,
6094                 "die if runtime exceeds this limit (in seconds)" },
6095         POPT_TABLEEND
6096 };
6097
6098 static int process_command(const struct ctdb_cmd *cmd, int argc,
6099                            const char **argv)
6100 {
6101         TALLOC_CTX *tmp_ctx;
6102         struct ctdb_context *ctdb;
6103         int ret;
6104         bool status;
6105         uint64_t srvid_offset;
6106
6107         tmp_ctx = talloc_new(NULL);
6108         if (tmp_ctx == NULL) {
6109                 fprintf(stderr, "Memory allocation error\n");
6110                 goto fail;
6111         }
6112
6113         if (cmd->without_daemon) {
6114                 if (options.pnn != -1) {
6115                         fprintf(stderr,
6116                                 "Cannot specify node for command %s\n",
6117                                 cmd->name);
6118                         goto fail;
6119                 }
6120
6121                 ret = cmd->fn(tmp_ctx, NULL, argc-1, argv+1);
6122                 talloc_free(tmp_ctx);
6123                 return ret;
6124         }
6125
6126         ctdb = talloc_zero(tmp_ctx, struct ctdb_context);
6127         if (ctdb == NULL) {
6128                 fprintf(stderr, "Memory allocation error\n");
6129                 goto fail;
6130         }
6131
6132         ctdb->ev = tevent_context_init(ctdb);
6133         if (ctdb->ev == NULL) {
6134                 fprintf(stderr, "Failed to initialize tevent\n");
6135                 goto fail;
6136         }
6137
6138         ret = ctdb_client_init(ctdb, ctdb->ev, options.socket, &ctdb->client);
6139         if (ret != 0) {
6140                 fprintf(stderr, "Failed to connect to CTDB daemon (%s)\n",
6141                         options.socket);
6142
6143                 if (!find_node_xpnn(ctdb, NULL)) {
6144                         fprintf(stderr, "Is this node part of CTDB cluster?\n");
6145                 }
6146                 goto fail;
6147         }
6148
6149         ctdb->pnn = ctdb_client_pnn(ctdb->client);
6150         srvid_offset = getpid() & 0xFFFF;
6151         ctdb->srvid = SRVID_CTDB_TOOL | (srvid_offset << 16);
6152
6153         if (options.pnn != -1) {
6154                 status = verify_pnn(ctdb, options.pnn);
6155                 if (! status) {
6156                         goto fail;
6157                 }
6158
6159                 ctdb->cmd_pnn = options.pnn;
6160         } else {
6161                 ctdb->cmd_pnn = ctdb->pnn;
6162         }
6163
6164         if (! cmd->remote && ctdb->pnn != ctdb->cmd_pnn) {
6165                 fprintf(stderr, "Node cannot be specified for command %s\n",
6166                         cmd->name);
6167                 goto fail;
6168         }
6169
6170         ret = cmd->fn(tmp_ctx, ctdb, argc-1, argv+1);
6171         talloc_free(tmp_ctx);
6172         return ret;
6173
6174 fail:
6175         talloc_free(tmp_ctx);
6176         return 1;
6177 }
6178
6179 static void signal_handler(int sig)
6180 {
6181         fprintf(stderr, "Maximum runtime exceeded - exiting\n");
6182 }
6183
6184 static void alarm_handler(int sig)
6185 {
6186         /* Kill any child processes */
6187         signal(SIGTERM, signal_handler);
6188         kill(0, SIGTERM);
6189
6190         _exit(1);
6191 }
6192
6193 int main(int argc, const char *argv[])
6194 {
6195         int opt;
6196         const char **extra_argv;
6197         int extra_argc;
6198         const struct ctdb_cmd *cmd;
6199         const char *ctdb_socket;
6200         int loglevel;
6201         int ret;
6202
6203         setlinebuf(stdout);
6204
6205         /* Set default options */
6206         options.socket = CTDB_SOCKET;
6207         options.debuglevelstr = NULL;
6208         options.timelimit = 10;
6209         options.sep = "|";
6210         options.maxruntime = 0;
6211         options.pnn = -1;
6212
6213         ctdb_socket = getenv("CTDB_SOCKET");
6214         if (ctdb_socket != NULL) {
6215                 options.socket = ctdb_socket;
6216         }
6217
6218         pc = poptGetContext(argv[0], argc, argv, cmdline_options,
6219                             POPT_CONTEXT_KEEP_FIRST);
6220         while ((opt = poptGetNextOpt(pc)) != -1) {
6221                 fprintf(stderr, "Invalid option %s: %s\n",
6222                         poptBadOption(pc, 0), poptStrerror(opt));
6223                 exit(1);
6224         }
6225
6226         if (options.maxruntime == 0) {
6227                 const char *ctdb_timeout;
6228
6229                 ctdb_timeout = getenv("CTDB_TIMEOUT");
6230                 if (ctdb_timeout != NULL) {
6231                         options.maxruntime = strtoul(ctdb_timeout, NULL, 0);
6232                 } else {
6233                         options.maxruntime = 120;
6234                 }
6235         }
6236         if (options.maxruntime <= 120) {
6237                 /* default timeout is 120 seconds */
6238                 options.maxruntime = 120;
6239         }
6240
6241         if (options.machineparsable) {
6242                 options.machinereadable = 1;
6243         }
6244
6245         /* setup the remaining options for the commands */
6246         extra_argc = 0;
6247         extra_argv = poptGetArgs(pc);
6248         if (extra_argv) {
6249                 extra_argv++;
6250                 while (extra_argv[extra_argc]) extra_argc++;
6251         }
6252
6253         if (extra_argc < 1) {
6254                 usage(NULL);
6255         }
6256
6257         cmd = match_command(extra_argv[0]);
6258         if (cmd == NULL) {
6259                 fprintf(stderr, "Unknown command '%s'\n", extra_argv[0]);
6260                 exit(1);
6261         }
6262
6263         /* Enable logging */
6264         setup_logging("ctdb", DEBUG_STDERR);
6265         if (debug_level_parse(options.debuglevelstr, &loglevel)) {
6266                 DEBUGLEVEL = loglevel;
6267         } else {
6268                 DEBUGLEVEL = DEBUG_ERR;
6269         }
6270
6271         signal(SIGALRM, alarm_handler);
6272         alarm(options.maxruntime);
6273
6274         ret = process_command(cmd, extra_argc, extra_argv);
6275         if (ret == -1) {
6276                 ret = 1;
6277         }
6278
6279         (void)poptFreeContext(pc);
6280
6281         return ret;
6282 }