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