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 == 1) {
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
3234                 req = process_clist_send(mem_ctx, ctdb, clist, count,
3235                                  ctdb_req_control_tcp_add_delayed_update,
3236                                  ctdb_reply_control_tcp_add_delayed_update);
3237                 if (req == NULL) {
3238                         return ENOMEM;
3239                 }
3240
3241                 tevent_req_poll(req, ctdb->ev);
3242
3243                 ret = process_clist_recv(req);
3244                 if (ret != 0) {
3245                         fprintf(stderr, "Failed to add %d tickles\n", ret);
3246                         return 1;
3247                 }
3248
3249                 return 0;
3250         }
3251
3252         if (! parse_ip_port(argv[0], &conn.src)) {
3253                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3254                 return 1;
3255         }
3256         if (! parse_ip_port(argv[1], &conn.dst)) {
3257                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3258                 return 1;
3259         }
3260
3261         ret = ctdb_ctrl_tcp_add_delayed_update(mem_ctx, ctdb->ev,
3262                                                ctdb->client, ctdb->cmd_pnn,
3263                                                TIMEOUT(), &conn);
3264         if (ret != 0) {
3265                 fprintf(stderr, "Failed to register connection\n");
3266                 return ret;
3267         }
3268
3269         return 0;
3270 }
3271
3272 static int control_deltickle(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3273                              int argc, const char **argv)
3274 {
3275         struct ctdb_connection conn;
3276         int ret;
3277
3278         if (argc != 0 && argc != 2) {
3279                 usage("deltickle");
3280         }
3281
3282         if (argc == 0) {
3283                 struct ctdb_connection *clist;
3284                 struct tevent_req *req;
3285                 int count;
3286
3287                 ret = ctdb_parse_connections(stdin, mem_ctx, &count, &clist);
3288                 if (ret != 0) {
3289                         return ret;
3290                 }
3291
3292                 req = process_clist_send(mem_ctx, ctdb, clist, count,
3293                                          ctdb_req_control_tcp_remove,
3294                                          ctdb_reply_control_tcp_remove);
3295                 if (req == NULL) {
3296                         return ENOMEM;
3297                 }
3298
3299                 tevent_req_poll(req, ctdb->ev);
3300
3301                 ret = process_clist_recv(req);
3302                 if (ret != 0) {
3303                         fprintf(stderr, "Failed to remove %d tickles\n", ret);
3304                         return 1;
3305                 }
3306
3307                 return 0;
3308         }
3309
3310         if (! parse_ip_port(argv[0], &conn.src)) {
3311                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3312                 return 1;
3313         }
3314         if (! parse_ip_port(argv[1], &conn.dst)) {
3315                 fprintf(stderr, "Invalid IP address %s\n", argv[1]);
3316                 return 1;
3317         }
3318
3319         ret = ctdb_ctrl_tcp_remove(mem_ctx, ctdb->ev, ctdb->client,
3320                                    ctdb->cmd_pnn, TIMEOUT(), &conn);
3321         if (ret != 0) {
3322                 fprintf(stderr, "Failed to unregister connection\n");
3323                 return ret;
3324         }
3325
3326         return 0;
3327 }
3328
3329 static int control_check_srvids(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3330                                 int argc, const char **argv)
3331 {
3332         uint64_t *srvid;
3333         uint8_t *result;
3334         int ret, i;
3335
3336         if (argc == 0) {
3337                 usage("check_srvids");
3338         }
3339
3340         srvid = talloc_array(mem_ctx, uint64_t, argc);
3341         if (srvid == NULL) {
3342                 fprintf(stderr, "Memory allocation error\n");
3343                 return 1;
3344         }
3345
3346         for (i=0; i<argc; i++) {
3347                 srvid[i] = strtoull(argv[i], NULL, 0);
3348         }
3349
3350         ret = ctdb_ctrl_check_srvids(mem_ctx, ctdb->ev, ctdb->client,
3351                                      ctdb->cmd_pnn, TIMEOUT(), srvid, argc,
3352                                      &result);
3353         if (ret != 0) {
3354                 fprintf(stderr, "Failed to check srvids on node %u\n",
3355                         ctdb->cmd_pnn);
3356                 return ret;
3357         }
3358
3359         for (i=0; i<argc; i++) {
3360                 printf("SRVID 0x%" PRIx64 " %s\n", srvid[i],
3361                        (result[i] ? "exists" : "does not exist"));
3362         }
3363
3364         return 0;
3365 }
3366
3367 static int control_listnodes(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3368                              int argc, const char **argv)
3369 {
3370         struct ctdb_node_map *nodemap;
3371         int i;
3372
3373         if (argc != 0) {
3374                 usage("listnodes");
3375         }
3376
3377         nodemap = read_nodes_file(mem_ctx, CTDB_UNKNOWN_PNN);
3378         if (nodemap == NULL) {
3379                 return 1;
3380         }
3381
3382         for (i=0; i<nodemap->num; i++) {
3383                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
3384                         continue;
3385                 }
3386
3387                 if (options.machinereadable) {
3388                         printf("%s%u%s%s%s\n", options.sep,
3389                                nodemap->node[i].pnn, options.sep,
3390                                ctdb_sock_addr_to_string(
3391                                         mem_ctx, &nodemap->node[i].addr),
3392                                options.sep);
3393                 } else {
3394                         printf("%s\n",
3395                                ctdb_sock_addr_to_string(
3396                                         mem_ctx, &nodemap->node[i].addr));
3397                 }
3398         }
3399
3400         return 0;
3401 }
3402
3403 static bool nodemap_identical(struct ctdb_node_map *nodemap1,
3404                               struct ctdb_node_map *nodemap2)
3405 {
3406         int i;
3407
3408         if (nodemap1->num != nodemap2->num) {
3409                 return false;
3410         }
3411
3412         for (i=0; i<nodemap1->num; i++) {
3413                 struct ctdb_node_and_flags *n1, *n2;
3414
3415                 n1 = &nodemap1->node[i];
3416                 n2 = &nodemap2->node[i];
3417
3418                 if ((n1->pnn != n2->pnn) ||
3419                     (n1->flags != n2->flags) ||
3420                     ! ctdb_sock_addr_same_ip(&n1->addr, &n2->addr)) {
3421                         return false;
3422                 }
3423         }
3424
3425         return true;
3426 }
3427
3428 static int check_node_file_changes(TALLOC_CTX *mem_ctx,
3429                                    struct ctdb_node_map *nm,
3430                                    struct ctdb_node_map *fnm,
3431                                    bool *reload)
3432 {
3433         int i;
3434         bool check_failed = false;
3435
3436         *reload = false;
3437
3438         for (i=0; i<nm->num; i++) {
3439                 if (i >= fnm->num) {
3440                         fprintf(stderr,
3441                                 "Node %u (%s) missing from nodes file\n",
3442                                 nm->node[i].pnn,
3443                                 ctdb_sock_addr_to_string(
3444                                         mem_ctx, &nm->node[i].addr));
3445                         check_failed = true;
3446                         continue;
3447                 }
3448                 if (nm->node[i].flags & NODE_FLAGS_DELETED &&
3449                     fnm->node[i].flags & NODE_FLAGS_DELETED) {
3450                         /* Node remains deleted */
3451                         continue;
3452                 }
3453
3454                 if (! (nm->node[i].flags & NODE_FLAGS_DELETED) &&
3455                     ! (fnm->node[i].flags & NODE_FLAGS_DELETED)) {
3456                         /* Node not newly nor previously deleted */
3457                         if (! ctdb_same_ip(&nm->node[i].addr,
3458                                            &fnm->node[i].addr)) {
3459                                 fprintf(stderr,
3460                                         "Node %u has changed IP address"
3461                                         " (was %s, now %s)\n",
3462                                         nm->node[i].pnn,
3463                                         ctdb_sock_addr_to_string(
3464                                                 mem_ctx, &nm->node[i].addr),
3465                                         ctdb_sock_addr_to_string(
3466                                                 mem_ctx, &fnm->node[i].addr));
3467                                 check_failed = true;
3468                         } else {
3469                                 if (nm->node[i].flags & NODE_FLAGS_DISCONNECTED) {
3470                                         fprintf(stderr,
3471                                                 "WARNING: Node %u is disconnected."
3472                                                 " You MUST fix this node manually!\n",
3473                                                 nm->node[i].pnn);
3474                                 }
3475                         }
3476                         continue;
3477                 }
3478
3479                 if (fnm->node[i].flags & NODE_FLAGS_DELETED) {
3480                         /* Node is being deleted */
3481                         printf("Node %u is DELETED\n", nm->node[i].pnn);
3482                         *reload = true;
3483                         if (! (nm->node[i].flags & NODE_FLAGS_DISCONNECTED)) {
3484                                 fprintf(stderr,
3485                                         "ERROR: Node %u is still connected\n",
3486                                         nm->node[i].pnn);
3487                                 check_failed = true;
3488                         }
3489                         continue;
3490                 }
3491
3492                 if (nm->node[i].flags & NODE_FLAGS_DELETED) {
3493                         /* Node was previously deleted */
3494                         printf("Node %u is UNDELETED\n", nm->node[i].pnn);
3495                         *reload = true;
3496                 }
3497         }
3498
3499         if (check_failed) {
3500                 fprintf(stderr,
3501                         "ERROR: Nodes will not be reloaded due to previous error\n");
3502                 return 1;
3503         }
3504
3505         /* Leftover nodes in file are NEW */
3506         for (; i < fnm->num; i++) {
3507                 printf("Node %u is NEW\n", fnm->node[i].pnn);
3508                 *reload = true;
3509         }
3510
3511         return 0;
3512 }
3513
3514 struct disable_recoveries_state {
3515         uint32_t *pnn_list;
3516         int node_count;
3517         bool *reply;
3518         int status;
3519         bool done;
3520 };
3521
3522 static void disable_recoveries_handler(uint64_t srvid, TDB_DATA data,
3523                                        void *private_data)
3524 {
3525         struct disable_recoveries_state *state =
3526                 (struct disable_recoveries_state *)private_data;
3527         int ret, i;
3528
3529         if (data.dsize != sizeof(int)) {
3530                 /* Ignore packet */
3531                 return;
3532         }
3533
3534         /* ret will be a PNN (i.e. >=0) on success, or negative on error */
3535         ret = *(int *)data.dptr;
3536         if (ret < 0) {
3537                 state->status = ret;
3538                 state->done = true;
3539                 return;
3540         }
3541         for (i=0; i<state->node_count; i++) {
3542                 if (state->pnn_list[i] == ret) {
3543                         state->reply[i] = true;
3544                         break;
3545                 }
3546         }
3547
3548         state->done = true;
3549         for (i=0; i<state->node_count; i++) {
3550                 if (! state->reply[i]) {
3551                         state->done = false;
3552                         break;
3553                 }
3554         }
3555 }
3556
3557 static int disable_recoveries(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3558                               uint32_t timeout, uint32_t *pnn_list, int count)
3559 {
3560         struct ctdb_disable_message disable = { 0 };
3561         struct disable_recoveries_state state;
3562         int ret, i;
3563
3564         disable.pnn = ctdb->pnn;
3565         disable.srvid = next_srvid(ctdb);
3566         disable.timeout = timeout;
3567
3568         state.pnn_list = pnn_list;
3569         state.node_count = count;
3570         state.done = false;
3571         state.status = 0;
3572         state.reply = talloc_zero_array(mem_ctx, bool, count);
3573         if (state.reply == NULL) {
3574                 return ENOMEM;
3575         }
3576
3577         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
3578                                               disable.srvid,
3579                                               disable_recoveries_handler,
3580                                               &state);
3581         if (ret != 0) {
3582                 return ret;
3583         }
3584
3585         for (i=0; i<count; i++) {
3586                 ret = ctdb_message_disable_recoveries(mem_ctx, ctdb->ev,
3587                                                       ctdb->client,
3588                                                       pnn_list[i],
3589                                                       &disable);
3590                 if (ret != 0) {
3591                         goto fail;
3592                 }
3593         }
3594
3595         ret = ctdb_client_wait_timeout(ctdb->ev, &state.done, TIMEOUT());
3596         if (ret == ETIME) {
3597                 fprintf(stderr, "Timed out waiting to disable recoveries\n");
3598         } else {
3599                 ret = (state.status >= 0 ? 0 : 1);
3600         }
3601
3602 fail:
3603         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
3604                                            disable.srvid, &state);
3605         return ret;
3606 }
3607
3608 static int control_reloadnodes(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3609                                int argc, const char **argv)
3610 {
3611         struct ctdb_node_map *nodemap = NULL;
3612         struct ctdb_node_map *file_nodemap;
3613         struct ctdb_node_map *remote_nodemap;
3614         struct ctdb_req_control request;
3615         struct ctdb_reply_control **reply;
3616         bool reload;
3617         int ret, i;
3618         uint32_t *pnn_list;
3619         int count;
3620
3621         nodemap = get_nodemap(ctdb, false);
3622         if (nodemap == NULL) {
3623                 return 1;
3624         }
3625
3626         file_nodemap = read_nodes_file(mem_ctx, ctdb->pnn);
3627         if (file_nodemap == NULL) {
3628                 return 1;
3629         }
3630
3631         for (i=0; i<nodemap->num; i++) {
3632                 if (nodemap->node[i].flags & NODE_FLAGS_DISCONNECTED) {
3633                         continue;
3634                 }
3635
3636                 ret = ctdb_ctrl_get_nodes_file(mem_ctx, ctdb->ev, ctdb->client,
3637                                                nodemap->node[i].pnn, TIMEOUT(),
3638                                                &remote_nodemap);
3639                 if (ret != 0) {
3640                         fprintf(stderr,
3641                                 "ERROR: Failed to get nodes file from node %u\n",
3642                                 nodemap->node[i].pnn);
3643                         return ret;
3644                 }
3645
3646                 if (! nodemap_identical(file_nodemap, remote_nodemap)) {
3647                         fprintf(stderr,
3648                                 "ERROR: Nodes file on node %u differs"
3649                                  " from current node (%u)\n",
3650                                  nodemap->node[i].pnn, ctdb->pnn);
3651                         return 1;
3652                 }
3653         }
3654
3655         ret = check_node_file_changes(mem_ctx, nodemap, file_nodemap, &reload);
3656         if (ret != 0) {
3657                 return ret;
3658         }
3659
3660         if (! reload) {
3661                 fprintf(stderr, "No change in nodes file,"
3662                                 " skipping unnecessary reload\n");
3663                 return 0;
3664         }
3665
3666         count = list_of_connected_nodes(nodemap, CTDB_UNKNOWN_PNN,
3667                                         mem_ctx, &pnn_list);
3668         if (count <= 0) {
3669                 fprintf(stderr, "Memory allocation error\n");
3670                 return 1;
3671         }
3672
3673         ret = disable_recoveries(mem_ctx, ctdb, 2*options.timelimit,
3674                                  pnn_list, count);
3675         if (ret != 0) {
3676                 fprintf(stderr, "Failed to disable recoveries\n");
3677                 return ret;
3678         }
3679
3680         ctdb_req_control_reload_nodes_file(&request);
3681         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
3682                                         pnn_list, count, TIMEOUT(),
3683                                         &request, NULL, &reply);
3684         if (ret != 0) {
3685                 bool failed = false;
3686
3687                 for (i=0; i<count; i++) {
3688                         ret = ctdb_reply_control_reload_nodes_file(reply[i]);
3689                         if (ret != 0) {
3690                                 fprintf(stderr,
3691                                         "Node %u failed to reload nodes\n",
3692                                         pnn_list[i]);
3693                                 failed = true;
3694                         }
3695                 }
3696                 if (failed) {
3697                         fprintf(stderr,
3698                                 "You MUST fix failed nodes manually!\n");
3699                 }
3700         }
3701
3702         ret = disable_recoveries(mem_ctx, ctdb, 0, pnn_list, count);
3703         if (ret != 0) {
3704                 fprintf(stderr, "Failed to enable recoveries\n");
3705                 return ret;
3706         }
3707
3708         return 0;
3709 }
3710
3711 static int moveip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3712                   ctdb_sock_addr *addr, uint32_t pnn)
3713 {
3714         struct ctdb_public_ip_list *pubip_list;
3715         struct ctdb_public_ip pubip;
3716         struct ctdb_node_map *nodemap;
3717         struct ctdb_req_control request;
3718         uint32_t *pnn_list;
3719         int ret, i, count;
3720
3721         ret = ctdb_message_disable_ip_check(mem_ctx, ctdb->ev, ctdb->client,
3722                                             CTDB_BROADCAST_CONNECTED,
3723                                             2*options.timelimit);
3724         if (ret != 0) {
3725                 fprintf(stderr, "Failed to disable IP check\n");
3726                 return ret;
3727         }
3728
3729         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3730                                        pnn, TIMEOUT(), &pubip_list);
3731         if (ret != 0) {
3732                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3733                         pnn);
3734                 return ret;
3735         }
3736
3737         for (i=0; i<pubip_list->num; i++) {
3738                 if (ctdb_same_ip(addr, &pubip_list->ip[i].addr)) {
3739                         break;
3740                 }
3741         }
3742
3743         if (i == pubip_list->num) {
3744                 fprintf(stderr, "Node %u CANNOT host IP address %s\n",
3745                         pnn, ctdb_sock_addr_to_string(mem_ctx, addr));
3746                 return 1;
3747         }
3748
3749         nodemap = get_nodemap(ctdb, false);
3750         if (nodemap == NULL) {
3751                 return 1;
3752         }
3753
3754         count = list_of_active_nodes(nodemap, pnn, mem_ctx, &pnn_list);
3755         if (count <= 0) {
3756                 fprintf(stderr, "Memory allocation error\n");
3757                 return 1;
3758         }
3759
3760         pubip.pnn = pnn;
3761         pubip.addr = *addr;
3762         ctdb_req_control_release_ip(&request, &pubip);
3763
3764         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
3765                                         pnn_list, count, TIMEOUT(),
3766                                         &request, NULL, NULL);
3767         if (ret != 0) {
3768                 fprintf(stderr, "Failed to release IP on nodes\n");
3769                 return ret;
3770         }
3771
3772         ret = ctdb_ctrl_takeover_ip(mem_ctx, ctdb->ev, ctdb->client,
3773                                     pnn, TIMEOUT(), &pubip);
3774         if (ret != 0) {
3775                 fprintf(stderr, "Failed to takeover IP on node %u\n", pnn);
3776                 return ret;
3777         }
3778
3779         return 0;
3780 }
3781
3782 static int control_moveip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3783                           int argc, const char **argv)
3784 {
3785         ctdb_sock_addr addr;
3786         uint32_t pnn;
3787         int ret, retries = 0;
3788
3789         if (argc != 2) {
3790                 usage("moveip");
3791         }
3792
3793         if (! parse_ip(argv[0], NULL, 0, &addr)) {
3794                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3795                 return 1;
3796         }
3797
3798         pnn = strtoul(argv[1], NULL, 10);
3799         if (pnn == CTDB_UNKNOWN_PNN) {
3800                 fprintf(stderr, "Invalid PNN %s\n", argv[1]);
3801                 return 1;
3802         }
3803
3804         while (retries < 5) {
3805                 ret = moveip(mem_ctx, ctdb, &addr, pnn);
3806                 if (ret == 0) {
3807                         break;
3808                 }
3809
3810                 sleep(3);
3811                 retries++;
3812         }
3813
3814         if (ret != 0) {
3815                 fprintf(stderr, "Failed to move IP %s to node %u\n",
3816                         argv[0], pnn);
3817                 return ret;
3818         }
3819
3820         return 0;
3821 }
3822
3823 static int control_rebalanceip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3824                                int argc, const char **argv)
3825 {
3826         ctdb_sock_addr addr;
3827         struct ctdb_node_map *nodemap;
3828         struct ctdb_public_ip pubip;
3829         struct ctdb_req_control request;
3830         uint32_t *pnn_list;
3831         int ret, count;
3832
3833         if (argc != 1) {
3834                 usage("rebalanceip");
3835         }
3836
3837         if (parse_ip(argv[0], NULL, 0, &addr)) {
3838                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3839                 return 1;
3840         }
3841
3842         ret = ctdb_message_disable_ip_check(mem_ctx, ctdb->ev, ctdb->client,
3843                                             CTDB_BROADCAST_CONNECTED,
3844                                             2*options.timelimit);
3845         if (ret != 0) {
3846                 fprintf(stderr, "Failed to disable IP check\n");
3847                 return 1;
3848         }
3849
3850         nodemap = get_nodemap(ctdb, false);
3851         if (nodemap == NULL) {
3852                 return 1;
3853         }
3854
3855         count = list_of_active_nodes(nodemap, -1, mem_ctx, &pnn_list);
3856         if (count <= 0) {
3857                 fprintf(stderr, "Memory allocation error\n");
3858                 return 1;
3859         }
3860
3861         pubip.pnn = CTDB_UNKNOWN_PNN;
3862         pubip.addr = addr;
3863
3864         ctdb_req_control_release_ip(&request, &pubip);
3865         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
3866                                         pnn_list, count, TIMEOUT(),
3867                                         &request, NULL, NULL);
3868         if (ret != 0) {
3869                 fprintf(stderr, "Failed to release IP from nodes\n");
3870                 return 1;
3871         }
3872
3873         return ipreallocate(mem_ctx, ctdb);
3874 }
3875
3876 static int rebalancenode(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3877                          uint32_t pnn)
3878 {
3879         int ret;
3880
3881         ret = ctdb_message_rebalance_node(mem_ctx, ctdb->ev, ctdb->client,
3882                                           CTDB_BROADCAST_CONNECTED, pnn);
3883         if (ret != 0) {
3884                 fprintf(stderr,
3885                         "Failed to ask recovery master to distribute IPs\n");
3886                 return ret;
3887         }
3888
3889         return 0;
3890 }
3891
3892 static int control_addip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3893                          int argc, const char **argv)
3894 {
3895         ctdb_sock_addr addr;
3896         struct ctdb_public_ip_list *pubip_list;
3897         struct ctdb_addr_info addr_info;
3898         unsigned int mask;
3899         int ret, i, retries = 0;
3900
3901         if (argc != 2) {
3902                 usage("addip");
3903         }
3904
3905         if (! parse_ip_mask(argv[0], argv[1], &addr, &mask)) {
3906                 fprintf(stderr, "Invalid IP/Mask %s\n", argv[0]);
3907                 return 1;
3908         }
3909
3910         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3911                                        ctdb->cmd_pnn, TIMEOUT(), &pubip_list);
3912         if (ret != 0) {
3913                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3914                         ctdb->cmd_pnn);
3915                 return 1;
3916         }
3917
3918         for (i=0; i<pubip_list->num; i++) {
3919                 if (ctdb_same_ip(&addr, &pubip_list->ip[i].addr)) {
3920                         fprintf(stderr, "Node already knows about IP %s\n",
3921                                 ctdb_sock_addr_to_string(mem_ctx, &addr));
3922                         return 0;
3923                 }
3924         }
3925
3926         addr_info.addr = addr;
3927         addr_info.mask = mask;
3928         addr_info.iface = argv[1];
3929
3930         while (retries < 5) {
3931                 ret = ctdb_ctrl_add_public_ip(mem_ctx, ctdb->ev, ctdb->client,
3932                                               ctdb->cmd_pnn, TIMEOUT(),
3933                                               &addr_info);
3934                 if (ret == 0) {
3935                         break;
3936                 }
3937
3938                 sleep(3);
3939                 retries++;
3940         }
3941
3942         if (ret != 0) {
3943                 fprintf(stderr, "Failed to add public IP to node %u."
3944                                 " Giving up\n", ctdb->cmd_pnn);
3945                 return ret;
3946         }
3947
3948         ret = rebalancenode(mem_ctx, ctdb, ctdb->cmd_pnn);
3949         if (ret != 0) {
3950                 return ret;
3951         }
3952
3953         return ipreallocate(mem_ctx, ctdb);
3954 }
3955
3956 static int control_delip(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
3957                          int argc, const char **argv)
3958 {
3959         ctdb_sock_addr addr;
3960         struct ctdb_public_ip_list *pubip_list;
3961         struct ctdb_addr_info addr_info;
3962         int ret, i;
3963
3964         if (argc != 1) {
3965                 usage("delip");
3966         }
3967
3968         if (! parse_ip(argv[0], NULL, 0, &addr)) {
3969                 fprintf(stderr, "Invalid IP address %s\n", argv[0]);
3970                 return 1;
3971         }
3972
3973         ret = ctdb_ctrl_get_public_ips(mem_ctx, ctdb->ev, ctdb->client,
3974                                        ctdb->cmd_pnn, TIMEOUT(), &pubip_list);
3975         if (ret != 0) {
3976                 fprintf(stderr, "Failed to get Public IPs from node %u\n",
3977                         ctdb->cmd_pnn);
3978                 return 1;
3979         }
3980
3981         for (i=0; i<pubip_list->num; i++) {
3982                 if (ctdb_same_ip(&addr, &pubip_list->ip[i].addr)) {
3983                         break;
3984                 }
3985         }
3986
3987         if (i == pubip_list->num) {
3988                 fprintf(stderr, "Node does not know about IP address %s\n",
3989                         ctdb_sock_addr_to_string(mem_ctx, &addr));
3990                 return 0;
3991         }
3992
3993         addr_info.addr = addr;
3994         addr_info.mask = 0;
3995         addr_info.iface = NULL;
3996
3997         ret = ctdb_ctrl_del_public_ip(mem_ctx, ctdb->ev, ctdb->client,
3998                                       ctdb->cmd_pnn, TIMEOUT(), &addr_info);
3999         if (ret != 0) {
4000                 fprintf(stderr, "Failed to delete public IP from node %u\n",
4001                         ctdb->cmd_pnn);
4002                 return ret;
4003         }
4004
4005         return 0;
4006 }
4007
4008 static int control_eventscript(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4009                                int argc, const char **argv)
4010 {
4011         int ret;
4012
4013         if (argc != 1) {
4014                 usage("eventscript");
4015         }
4016
4017         if (strcmp(argv[0], "monitor") != 0) {
4018                 fprintf(stderr, "Only monitor event can be run\n");
4019                 return 1;
4020         }
4021
4022         ret = ctdb_ctrl_run_eventscripts(mem_ctx, ctdb->ev, ctdb->client,
4023                                          ctdb->cmd_pnn, TIMEOUT(), argv[0]);
4024         if (ret != 0) {
4025                 fprintf(stderr, "Failed to run monitor event on node %u\n",
4026                         ctdb->cmd_pnn);
4027                 return ret;
4028         }
4029
4030         return 0;
4031 }
4032
4033 #define DB_VERSION      3
4034 #define MAX_DB_NAME     64
4035 #define MAX_REC_BUFFER_SIZE     (100*1000)
4036
4037 struct db_header {
4038         unsigned long version;
4039         time_t timestamp;
4040         unsigned long flags;
4041         unsigned long nbuf;
4042         unsigned long nrec;
4043         char name[MAX_DB_NAME];
4044 };
4045
4046 struct backup_state {
4047         TALLOC_CTX *mem_ctx;
4048         struct ctdb_rec_buffer *recbuf;
4049         uint32_t db_id;
4050         int fd;
4051         unsigned int nbuf, nrec;
4052 };
4053
4054 static int backup_handler(uint32_t reqid, struct ctdb_ltdb_header *header,
4055                           TDB_DATA key, TDB_DATA data, void *private_data)
4056 {
4057         struct backup_state *state = (struct backup_state *)private_data;
4058         size_t len;
4059         int ret;
4060
4061         if (state->recbuf == NULL) {
4062                 state->recbuf = ctdb_rec_buffer_init(state->mem_ctx,
4063                                                      state->db_id);
4064                 if (state->recbuf == NULL) {
4065                         return ENOMEM;
4066                 }
4067         }
4068
4069         ret = ctdb_rec_buffer_add(state->recbuf, state->recbuf, reqid,
4070                                   header, key, data);
4071         if (ret != 0) {
4072                 return ret;
4073         }
4074
4075         len = ctdb_rec_buffer_len(state->recbuf);
4076         if (len < MAX_REC_BUFFER_SIZE) {
4077                 return 0;
4078         }
4079
4080         ret = ctdb_rec_buffer_write(state->recbuf, state->fd);
4081         if (ret != 0) {
4082                 fprintf(stderr, "Failed to write records to backup file\n");
4083                 return ret;
4084         }
4085
4086         state->nbuf += 1;
4087         state->nrec += state->recbuf->count;
4088         TALLOC_FREE(state->recbuf);
4089
4090         return 0;
4091 }
4092
4093 static int control_backupdb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4094                             int argc, const char **argv)
4095 {
4096         const char *db_name;
4097         struct ctdb_db_context *db;
4098         uint32_t db_id;
4099         uint8_t db_flags;
4100         struct backup_state state;
4101         struct db_header db_hdr;
4102         int fd, ret;
4103
4104         if (argc != 2) {
4105                 usage("backupdb");
4106         }
4107
4108         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
4109                 return 1;
4110         }
4111
4112         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4113                           db_flags, &db);
4114         if (ret != 0) {
4115                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4116                 return ret;
4117         }
4118
4119         fd = open(argv[1], O_RDWR|O_CREAT, 0600);
4120         if (fd == -1) {
4121                 ret = errno;
4122                 fprintf(stderr, "Failed to open file %s for writing\n",
4123                         argv[1]);
4124                 return ret;
4125         }
4126
4127         /* Write empty header first */
4128         ZERO_STRUCT(db_hdr);
4129         ret = write(fd, &db_hdr, sizeof(struct db_header));
4130         if (ret == -1) {
4131                 ret = errno;
4132                 close(fd);
4133                 fprintf(stderr, "Failed to write header to file %s\n", argv[1]);
4134                 return ret;
4135         }
4136
4137         state.mem_ctx = mem_ctx;
4138         state.recbuf = NULL;
4139         state.fd = fd;
4140         state.nbuf = 0;
4141         state.nrec = 0;
4142
4143         ret = ctdb_db_traverse(db, true, false, backup_handler, &state);
4144         if (ret != 0) {
4145                 fprintf(stderr, "Failed to collect records from DB %s\n",
4146                         db_name);
4147                 close(fd);
4148                 return ret;
4149         }
4150
4151         if (state.recbuf != NULL) {
4152                 ret = ctdb_rec_buffer_write(state.recbuf, state.fd);
4153                 if (ret != 0) {
4154                         fprintf(stderr,
4155                                 "Failed to write records to backup file\n");
4156                         close(fd);
4157                         return ret;
4158                 }
4159
4160                 state.nbuf += 1;
4161                 state.nrec += state.recbuf->count;
4162                 TALLOC_FREE(state.recbuf);
4163         }
4164
4165         db_hdr.version = DB_VERSION;
4166         db_hdr.timestamp = time(NULL);
4167         db_hdr.flags = db_flags;
4168         db_hdr.nbuf = state.nbuf;
4169         db_hdr.nrec = state.nrec;
4170         strncpy(db_hdr.name, db_name, MAX_DB_NAME-1);
4171
4172         lseek(fd, 0, SEEK_SET);
4173         ret = write(fd, &db_hdr, sizeof(struct db_header));
4174         if (ret == -1) {
4175                 ret = errno;
4176                 close(fd);
4177                 fprintf(stderr, "Failed to write header to file %s\n", argv[1]);
4178                 return ret;
4179         }
4180
4181         close(fd);
4182         printf("Database backed up to %s\n", argv[1]);
4183         return 0;
4184 }
4185
4186 static int control_restoredb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4187                              int argc, const char **argv)
4188 {
4189         const char *db_name = NULL;
4190         struct ctdb_db_context *db;
4191         struct db_header db_hdr;
4192         struct ctdb_node_map *nodemap;
4193         struct ctdb_req_control request;
4194         struct ctdb_reply_control **reply;
4195         struct ctdb_transdb wipedb;
4196         struct ctdb_pulldb_ext pulldb;
4197         struct ctdb_rec_buffer *recbuf;
4198         uint32_t generation;
4199         uint32_t *pnn_list;
4200         char timebuf[128];
4201         int fd, i;
4202         int count, ret;
4203
4204         if (argc < 1 || argc > 2) {
4205                 usage("restoredb");
4206         }
4207
4208         fd = open(argv[0], O_RDONLY, 0600);
4209         if (fd == -1) {
4210                 ret = errno;
4211                 fprintf(stderr, "Failed to open file %s for reading\n",
4212                         argv[0]);
4213                 return ret;
4214         }
4215
4216         if (argc == 2) {
4217                 db_name = argv[1];
4218         }
4219
4220         ret = read(fd, &db_hdr, sizeof(struct db_header));
4221         if (ret == -1) {
4222                 ret = errno;
4223                 close(fd);
4224                 fprintf(stderr, "Failed to read db header from file %s\n",
4225                         argv[0]);
4226                 return ret;
4227         }
4228
4229         if (db_hdr.version != DB_VERSION) {
4230                 fprintf(stderr,
4231                         "Wrong version of backup file, expected %u, got %lu\n",
4232                         DB_VERSION, db_hdr.version);
4233                 close(fd);
4234                 return EINVAL;
4235         }
4236
4237         if (db_name == NULL) {
4238                 db_name = db_hdr.name;
4239         }
4240
4241         strftime(timebuf, sizeof(timebuf)-1, "%Y/%m/%d %H:%M:%S",
4242                  localtime(&db_hdr.timestamp));
4243         printf("Restoring database %s from backup @ %s\n", db_name, timebuf);
4244
4245         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4246                           db_hdr.flags, &db);
4247         if (ret != 0) {
4248                 fprintf(stderr, "Failed to attach to DB %s\n", db_hdr.name);
4249                 return ret;
4250         }
4251
4252         nodemap = get_nodemap(ctdb, false);
4253         if (nodemap == NULL) {
4254                 fprintf(stderr, "Failed to get nodemap\n");
4255                 close(fd);
4256                 return ENOMEM;
4257         }
4258
4259         ret = get_generation(mem_ctx, ctdb, &generation);
4260         if (ret != 0) {
4261                 fprintf(stderr, "Failed to get current generation\n");
4262                 close(fd);
4263                 return ret;
4264         }
4265
4266         count = list_of_active_nodes(nodemap, CTDB_UNKNOWN_PNN, mem_ctx,
4267                                      &pnn_list);
4268         if (count <= 0) {
4269                 close(fd);
4270                 return ENOMEM;
4271         }
4272
4273         wipedb.db_id = ctdb_db_id(db);
4274         wipedb.tid = generation;
4275
4276         ctdb_req_control_db_freeze(&request, wipedb.db_id);
4277         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4278                                         ctdb->client, pnn_list, count,
4279                                         TIMEOUT(), &request, NULL, NULL);
4280         if (ret != 0) {
4281                 goto failed;
4282         }
4283
4284
4285         ctdb_req_control_db_transaction_start(&request, &wipedb);
4286         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4287                                         pnn_list, count, TIMEOUT(),
4288                                         &request, NULL, NULL);
4289         if (ret != 0) {
4290                 goto failed;
4291         }
4292
4293         ctdb_req_control_wipe_database(&request, &wipedb);
4294         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4295                                         pnn_list, count, TIMEOUT(),
4296                                         &request, NULL, NULL);
4297         if (ret != 0) {
4298                 goto failed;
4299         }
4300
4301         pulldb.db_id = ctdb_db_id(db);
4302         pulldb.lmaster = 0;
4303         pulldb.srvid = SRVID_CTDB_PUSHDB;
4304
4305         ctdb_req_control_db_push_start(&request, &pulldb);
4306         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4307                                         pnn_list, count, TIMEOUT(),
4308                                         &request, NULL, NULL);
4309         if (ret != 0) {
4310                 goto failed;
4311         }
4312
4313         for (i=0; i<db_hdr.nbuf; i++) {
4314                 struct ctdb_req_message message;
4315                 TDB_DATA data;
4316
4317                 ret = ctdb_rec_buffer_read(fd, mem_ctx, &recbuf);
4318                 if (ret != 0) {
4319                         goto failed;
4320                 }
4321
4322                 data.dsize = ctdb_rec_buffer_len(recbuf);
4323                 data.dptr = talloc_size(mem_ctx, data.dsize);
4324                 if (data.dptr == NULL) {
4325                         goto failed;
4326                 }
4327
4328                 ctdb_rec_buffer_push(recbuf, data.dptr);
4329
4330                 message.srvid = pulldb.srvid;
4331                 message.data.data = data;
4332
4333                 ret = ctdb_client_message_multi(mem_ctx, ctdb->ev,
4334                                                 ctdb->client,
4335                                                 pnn_list, count,
4336                                                 &message, NULL);
4337                 if (ret != 0) {
4338                         goto failed;
4339                 }
4340
4341                 talloc_free(recbuf);
4342                 talloc_free(data.dptr);
4343         }
4344
4345         ctdb_req_control_db_push_confirm(&request, pulldb.db_id);
4346         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4347                                         pnn_list, count, TIMEOUT(),
4348                                         &request, NULL, &reply);
4349         if (ret != 0) {
4350                 goto failed;
4351         }
4352
4353         for (i=0; i<count; i++) {
4354                 uint32_t num_records;
4355
4356                 ret = ctdb_reply_control_db_push_confirm(reply[i],
4357                                                          &num_records);
4358                 if (ret != 0) {
4359                         fprintf(stderr, "Invalid response from node %u\n",
4360                                 pnn_list[i]);
4361                         goto failed;
4362                 }
4363
4364                 if (num_records != db_hdr.nrec) {
4365                         fprintf(stderr, "Node %u received %u of %lu records\n",
4366                                 pnn_list[i], num_records, db_hdr.nrec);
4367                         goto failed;
4368                 }
4369         }
4370
4371         ctdb_req_control_db_set_healthy(&request, wipedb.db_id);
4372         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4373                                         pnn_list, count, TIMEOUT(),
4374                                         &request, NULL, NULL);
4375         if (ret != 0) {
4376                 goto failed;
4377         }
4378
4379         ctdb_req_control_db_transaction_commit(&request, &wipedb);
4380         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4381                                         pnn_list, count, TIMEOUT(),
4382                                         &request, NULL, NULL);
4383         if (ret != 0) {
4384                 goto failed;
4385         }
4386
4387         ctdb_req_control_db_thaw(&request, wipedb.db_id);
4388         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4389                                         ctdb->client, pnn_list, count,
4390                                         TIMEOUT(), &request, NULL, NULL);
4391         if (ret != 0) {
4392                 goto failed;
4393         }
4394
4395         printf("Database %s restored\n", db_name);
4396         return 0;
4397
4398
4399 failed:
4400         close(fd);
4401         ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
4402                               ctdb->pnn, TIMEOUT(), CTDB_RECOVERY_ACTIVE);
4403         return ret;
4404 }
4405
4406 struct dumpdbbackup_state {
4407         ctdb_rec_parser_func_t parser;
4408         struct dump_record_state sub_state;
4409 };
4410
4411 static int dumpdbbackup_handler(uint32_t reqid,
4412                                 struct ctdb_ltdb_header *header,
4413                                 TDB_DATA key, TDB_DATA data,
4414                                 void *private_data)
4415 {
4416         struct dumpdbbackup_state *state =
4417                 (struct dumpdbbackup_state *)private_data;
4418         struct ctdb_ltdb_header hdr;
4419         int ret;
4420
4421         ret = ctdb_ltdb_header_extract(&data, &hdr);
4422         if (ret != 0) {
4423                 return ret;
4424         }
4425
4426         return state->parser(reqid, &hdr, key, data, &state->sub_state);
4427 }
4428
4429 static int control_dumpdbbackup(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4430                                 int argc, const char **argv)
4431 {
4432         struct db_header db_hdr;
4433         char timebuf[128];
4434         struct dumpdbbackup_state state;
4435         int fd, ret, i;
4436
4437         if (argc != 1) {
4438                 usage("dumpbackup");
4439         }
4440
4441         fd = open(argv[0], O_RDONLY, 0600);
4442         if (fd == -1) {
4443                 ret = errno;
4444                 fprintf(stderr, "Failed to open file %s for reading\n",
4445                         argv[0]);
4446                 return ret;
4447         }
4448
4449         ret = read(fd, &db_hdr, sizeof(struct db_header));
4450         if (ret == -1) {
4451                 ret = errno;
4452                 close(fd);
4453                 fprintf(stderr, "Failed to read db header from file %s\n",
4454                         argv[0]);
4455                 return ret;
4456         }
4457
4458         if (db_hdr.version != DB_VERSION) {
4459                 fprintf(stderr,
4460                         "Wrong version of backup file, expected %u, got %lu\n",
4461                         DB_VERSION, db_hdr.version);
4462                 close(fd);
4463                 return EINVAL;
4464         }
4465
4466         strftime(timebuf, sizeof(timebuf)-1, "%Y/%m/%d %H:%M:%S",
4467                  localtime(&db_hdr.timestamp));
4468         printf("Dumping database %s from backup @ %s\n",
4469                db_hdr.name, timebuf);
4470
4471         state.parser = dump_record;
4472         state.sub_state.count = 0;
4473
4474         for (i=0; i<db_hdr.nbuf; i++) {
4475                 struct ctdb_rec_buffer *recbuf;
4476
4477                 ret = ctdb_rec_buffer_read(fd, mem_ctx, &recbuf);
4478                 if (ret != 0) {
4479                         fprintf(stderr, "Failed to read records\n");
4480                         close(fd);
4481                         return ret;
4482                 }
4483
4484                 ret = ctdb_rec_buffer_traverse(recbuf, dumpdbbackup_handler,
4485                                                &state);
4486                 if (ret != 0) {
4487                         fprintf(stderr, "Failed to dump records\n");
4488                         close(fd);
4489                         return ret;
4490                 }
4491         }
4492
4493         close(fd);
4494         printf("Dumped %u record(s)\n", state.sub_state.count);
4495         return 0;
4496 }
4497
4498 static int control_wipedb(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4499                           int argc, const char **argv)
4500 {
4501         const char *db_name;
4502         struct ctdb_db_context *db;
4503         uint32_t db_id;
4504         uint8_t db_flags;
4505         struct ctdb_node_map *nodemap;
4506         struct ctdb_req_control request;
4507         struct ctdb_transdb wipedb;
4508         uint32_t generation;
4509         uint32_t *pnn_list;
4510         int count, ret;
4511
4512         if (argc != 1) {
4513                 usage("wipedb");
4514         }
4515
4516         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, &db_flags)) {
4517                 return 1;
4518         }
4519
4520         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
4521                           db_flags, &db);
4522         if (ret != 0) {
4523                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
4524                 return ret;
4525         }
4526
4527         nodemap = get_nodemap(ctdb, false);
4528         if (nodemap == NULL) {
4529                 fprintf(stderr, "Failed to get nodemap\n");
4530                 return ENOMEM;
4531         }
4532
4533         ret = get_generation(mem_ctx, ctdb, &generation);
4534         if (ret != 0) {
4535                 fprintf(stderr, "Failed to get current generation\n");
4536                 return ret;
4537         }
4538
4539         count = list_of_active_nodes(nodemap, CTDB_UNKNOWN_PNN, mem_ctx,
4540                                      &pnn_list);
4541         if (count <= 0) {
4542                 return ENOMEM;
4543         }
4544
4545         ctdb_req_control_db_freeze(&request, db_id);
4546         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4547                                         ctdb->client, pnn_list, count,
4548                                         TIMEOUT(), &request, NULL, NULL);
4549         if (ret != 0) {
4550                 goto failed;
4551         }
4552
4553         wipedb.db_id = db_id;
4554         wipedb.tid = generation;
4555
4556         ctdb_req_control_db_transaction_start(&request, &wipedb);
4557         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4558                                         pnn_list, count, TIMEOUT(),
4559                                         &request, NULL, NULL);
4560         if (ret != 0) {
4561                 goto failed;
4562         }
4563
4564         ctdb_req_control_wipe_database(&request, &wipedb);
4565         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4566                                         pnn_list, count, TIMEOUT(),
4567                                         &request, NULL, NULL);
4568         if (ret != 0) {
4569                 goto failed;
4570         }
4571
4572         ctdb_req_control_db_set_healthy(&request, db_id);
4573         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4574                                         pnn_list, count, TIMEOUT(),
4575                                         &request, NULL, NULL);
4576         if (ret != 0) {
4577                 goto failed;
4578         }
4579
4580         ctdb_req_control_db_transaction_commit(&request, &wipedb);
4581         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
4582                                         pnn_list, count, TIMEOUT(),
4583                                         &request, NULL, NULL);
4584         if (ret != 0) {
4585                 goto failed;
4586         }
4587
4588         ctdb_req_control_db_thaw(&request, db_id);
4589         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev,
4590                                         ctdb->client, pnn_list, count,
4591                                         TIMEOUT(), &request, NULL, NULL);
4592         if (ret != 0) {
4593                 goto failed;
4594         }
4595
4596         printf("Database %s wiped\n", db_name);
4597         return 0;
4598
4599
4600 failed:
4601         ctdb_ctrl_set_recmode(mem_ctx, ctdb->ev, ctdb->client,
4602                               ctdb->pnn, TIMEOUT(), CTDB_RECOVERY_ACTIVE);
4603         return ret;
4604 }
4605
4606 static int control_recmaster(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4607                              int argc, const char **argv)
4608 {
4609         uint32_t recmaster;
4610         int ret;
4611
4612         ret = ctdb_ctrl_get_recmaster(mem_ctx, ctdb->ev, ctdb->client,
4613                                       ctdb->cmd_pnn, TIMEOUT(), &recmaster);
4614         if (ret != 0) {
4615                 return ret;
4616         }
4617
4618         printf("%u\n", recmaster);
4619         return 0;
4620 }
4621
4622 static void print_scriptstatus_one(struct ctdb_script_list *slist,
4623                                    const char *event_str)
4624 {
4625         int i;
4626         int num_run = 0;
4627
4628         if (slist == NULL) {
4629                 if (! options.machinereadable) {
4630                         printf("%s cycle never run\n", event_str);
4631                 }
4632                 return;
4633         }
4634
4635         for (i=0; i<slist->num_scripts; i++) {
4636                 if (slist->script[i].status != -ENOEXEC) {
4637                         num_run++;
4638                 }
4639         }
4640
4641         if (! options.machinereadable) {
4642                 printf("%d scripts were executed last %s cycle\n",
4643                        num_run, event_str);
4644         }
4645
4646         for (i=0; i<slist->num_scripts; i++) {
4647                 const char *status = NULL;
4648
4649                 switch (slist->script[i].status) {
4650                 case -ETIME:
4651                         status = "TIMEDOUT";
4652                         break;
4653                 case -ENOEXEC:
4654                         status = "DISABLED";
4655                         break;
4656                 case 0:
4657                         status = "OK";
4658                         break;
4659                 default:
4660                         if (slist->script[i].status > 0) {
4661                                 status = "ERROR";
4662                         }
4663                         break;
4664                 }
4665
4666                 if (options.machinereadable) {
4667                         printf("%s%s%s%s%s%i%s%s%s%lu.%06lu%s%lu.%06lu%s%s%s\n",
4668                                options.sep,
4669                                event_str, options.sep,
4670                                slist->script[i].name, options.sep,
4671                                slist->script[i].status, options.sep,
4672                                status, options.sep,
4673                                slist->script[i].start.tv_sec,
4674                                slist->script[i].start.tv_usec, options.sep,
4675                                slist->script[i].finished.tv_sec,
4676                                slist->script[i].finished.tv_usec, options.sep,
4677                                slist->script[i].output, options.sep);
4678                         continue;
4679                 }
4680
4681                 if (status) {
4682                         printf("%-20s Status:%s    ",
4683                                slist->script[i].name, status);
4684                 } else {
4685                         /* Some other error, eg from stat. */
4686                         printf("%-20s Status:CANNOT RUN (%s)",
4687                                slist->script[i].name,
4688                                strerror(-slist->script[i].status));
4689                 }
4690
4691                 if (slist->script[i].status >= 0) {
4692                         printf("Duration:%.3lf ",
4693                                timeval_delta(&slist->script[i].finished,
4694                                              &slist->script[i].start));
4695                 }
4696                 if (slist->script[i].status != -ENOEXEC) {
4697                         printf("%s", ctime(&slist->script[i].start.tv_sec));
4698                         if (slist->script[i].status != 0) {
4699                                 printf("   OUTPUT:%s\n",
4700                                        slist->script[i].output);
4701                         }
4702                 } else {
4703                         printf("\n");
4704                 }
4705         }
4706 }
4707
4708 static void print_scriptstatus(struct ctdb_script_list **slist,
4709                                int count, const char **event_str)
4710 {
4711         int i;
4712
4713         if (options.machinereadable) {
4714                 printf("%s%s%s%s%s%s%s%s%s%s%s%s%s%s%s\n",
4715                        options.sep,
4716                        "Type", options.sep,
4717                        "Name", options.sep,
4718                        "Code", options.sep,
4719                        "Status", options.sep,
4720                        "Start", options.sep,
4721                        "End", options.sep,
4722                        "Error Output", options.sep);
4723         }
4724
4725         for (i=0; i<count; i++) {
4726                 print_scriptstatus_one(slist[i], event_str[i]);
4727         }
4728 }
4729
4730 static int control_scriptstatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4731                                 int argc, const char **argv)
4732 {
4733         struct ctdb_script_list **slist;
4734         const char *event_str;
4735         enum ctdb_event event;
4736         const char *all_events[] = {
4737                 "init", "setup", "startup", "monitor",
4738                 "takeip", "releaseip", "updateip", "ipreallocated" };
4739         bool valid;
4740         int ret, i, j;
4741         int count, start, end, num;
4742
4743         if (argc > 1) {
4744                 usage("scriptstatus");
4745         }
4746
4747         if (argc == 0) {
4748                 event_str = "monitor";
4749         } else {
4750                 event_str = argv[0];
4751         }
4752
4753         valid = false;
4754         for (i=0; i<ARRAY_SIZE(all_events); i++) {
4755                 if (strcmp(event_str, all_events[i]) == 0) {
4756                         valid = true;
4757                         count = 1;
4758                         start = i;
4759                         end = i+1;
4760                         break;
4761                 }
4762         }
4763
4764         if (strcmp(event_str, "all") == 0) {
4765                 valid = true;
4766                 count = ARRAY_SIZE(all_events);
4767                 start = 0;
4768                 end = count-1;
4769         }
4770
4771         if (! valid) {
4772                 fprintf(stderr, "Unknown event name %s\n", argv[0]);
4773                 usage("scriptstatus");
4774         }
4775
4776         slist = talloc_array(mem_ctx, struct ctdb_script_list *, count);
4777         if (slist == NULL) {
4778                 fprintf(stderr, "Memory allocation error\n");
4779                 return 1;
4780         }
4781
4782         num = 0;
4783         for (i=start; i<end; i++) {
4784                 event = ctdb_event_from_string(all_events[i]);
4785
4786                 ret = ctdb_ctrl_get_event_script_status(mem_ctx, ctdb->ev,
4787                                                         ctdb->client,
4788                                                         ctdb->cmd_pnn,
4789                                                         TIMEOUT(), event,
4790                                                         &slist[num]);
4791                 if (ret != 0) {
4792                         fprintf(stderr,
4793                                 "failed to get script status for %s event\n",
4794                                 all_events[i]);
4795                         return 1;
4796                 }
4797
4798                 if (slist[num] == NULL) {
4799                         num++;
4800                         continue;
4801                 }
4802
4803                 /* The ETIME status is ignored for certain events.
4804                  * In that case the status is 0, but endtime is not set.
4805                  */
4806                 for (j=0; j<slist[num]->num_scripts; j++) {
4807                         if (slist[num]->script[j].status == 0 &&
4808                             timeval_is_zero(&slist[num]->script[j].finished)) {
4809                                 slist[num]->script[j].status = -ETIME;
4810                         }
4811                 }
4812
4813                 num++;
4814         }
4815
4816         print_scriptstatus(slist, count, &all_events[start]);
4817         return 0;
4818 }
4819
4820 static int control_enablescript(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4821                                 int argc, const char **argv)
4822 {
4823         int ret;
4824
4825         if (argc != 1) {
4826                 usage("enablescript");
4827         }
4828
4829         ret = ctdb_ctrl_enable_script(mem_ctx, ctdb->ev, ctdb->client,
4830                                       ctdb->cmd_pnn, TIMEOUT(), argv[0]);
4831         if (ret != 0) {
4832                 fprintf(stderr, "Failed to enable script %s\n", argv[0]);
4833                 return ret;
4834         }
4835
4836         return 0;
4837 }
4838
4839 static int control_disablescript(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4840                                  int argc, const char **argv)
4841 {
4842         int ret;
4843
4844         if (argc != 1) {
4845                 usage("disablescript");
4846         }
4847
4848         ret = ctdb_ctrl_disable_script(mem_ctx, ctdb->ev, ctdb->client,
4849                                        ctdb->cmd_pnn, TIMEOUT(), argv[0]);
4850         if (ret != 0) {
4851                 fprintf(stderr, "Failed to disable script %s\n", argv[0]);
4852                 return ret;
4853         }
4854
4855         return 0;
4856 }
4857
4858 static int control_natgw(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4859                          int argc, const char **argv)
4860 {
4861         char *t, *natgw_helper = NULL;
4862
4863         if (argc != 1) {
4864                 usage("natgw");
4865         }
4866
4867         t = getenv("CTDB_NATGW_HELPER");
4868         if (t != NULL) {
4869                 natgw_helper = talloc_strdup(mem_ctx, t);
4870         } else {
4871                 natgw_helper = talloc_asprintf(mem_ctx, "%s/ctdb_natgw",
4872                                                CTDB_HELPER_BINDIR);
4873         }
4874
4875         if (natgw_helper == NULL) {
4876                 fprintf(stderr, "Unable to set NAT gateway helper\n");
4877                 return 1;
4878         }
4879
4880         return run_helper("NAT gateway helper", natgw_helper, argv[0]);
4881 }
4882
4883 static int control_natgwlist(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4884                              int argc, const char **argv)
4885 {
4886         char *t, *natgw_helper = NULL;
4887
4888         if (argc != 0) {
4889                 usage("natgwlist");
4890         }
4891
4892         t = getenv("CTDB_NATGW_HELPER");
4893         if (t != NULL) {
4894                 natgw_helper = talloc_strdup(mem_ctx, t);
4895         } else {
4896                 natgw_helper = talloc_asprintf(mem_ctx, "%s/ctdb_natgw",
4897                                                CTDB_HELPER_BINDIR);
4898         }
4899
4900         if (natgw_helper == NULL) {
4901                 fprintf(stderr, "Unable to set NAT gateway helper\n");
4902                 return 1;
4903         }
4904
4905         return run_helper("NAT gateway helper", natgw_helper, "natgwlist");
4906 }
4907
4908 /*
4909  * Find the PNN of the current node
4910  * discover the pnn by loading the nodes file and try to bind
4911  * to all addresses one at a time until the ip address is found.
4912  */
4913 static bool find_node_xpnn(TALLOC_CTX *mem_ctx, uint32_t *pnn)
4914 {
4915         struct ctdb_node_map *nodemap;
4916         int i;
4917
4918         nodemap = read_nodes_file(mem_ctx, CTDB_UNKNOWN_PNN);
4919         if (nodemap == NULL) {
4920                 return false;
4921         }
4922
4923         for (i=0; i<nodemap->num; i++) {
4924                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
4925                         continue;
4926                 }
4927                 if (ctdb_sys_have_ip(&nodemap->node[i].addr)) {
4928                         if (pnn != NULL) {
4929                                 *pnn = nodemap->node[i].pnn;
4930                         }
4931                         talloc_free(nodemap);
4932                         return true;
4933                 }
4934         }
4935
4936         fprintf(stderr, "Failed to detect PNN of the current node.\n");
4937         talloc_free(nodemap);
4938         return false;
4939 }
4940
4941 static int control_getreclock(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
4942                               int argc, const char **argv)
4943 {
4944         const char *reclock;
4945         int ret;
4946
4947         if (argc != 0) {
4948                 usage("getreclock");
4949         }
4950
4951         ret = ctdb_ctrl_get_reclock_file(mem_ctx, ctdb->ev, ctdb->client,
4952                                          ctdb->cmd_pnn, TIMEOUT(), &reclock);
4953         if (ret != 0) {
4954                 return ret;
4955         }
4956
4957         if (reclock != NULL) {
4958                 printf("%s\n", reclock);
4959         }
4960
4961         return 0;
4962 }
4963
4964 static int control_setlmasterrole(TALLOC_CTX *mem_ctx,
4965                                   struct ctdb_context *ctdb,
4966                                   int argc, const char **argv)
4967 {
4968         uint32_t lmasterrole = 0;
4969         int ret;
4970
4971         if (argc != 1) {
4972                 usage("setlmasterrole");
4973         }
4974
4975         if (strcmp(argv[0], "on") == 0) {
4976                 lmasterrole = 1;
4977         } else if (strcmp(argv[0], "off") == 0) {
4978                 lmasterrole = 0;
4979         } else {
4980                 usage("setlmasterrole");
4981         }
4982
4983         ret = ctdb_ctrl_set_lmasterrole(mem_ctx, ctdb->ev, ctdb->client,
4984                                         ctdb->cmd_pnn, TIMEOUT(), lmasterrole);
4985         if (ret != 0) {
4986                 return ret;
4987         }
4988
4989         return 0;
4990 }
4991
4992 static int control_setrecmasterrole(TALLOC_CTX *mem_ctx,
4993                                     struct ctdb_context *ctdb,
4994                                     int argc, const char **argv)
4995 {
4996         uint32_t recmasterrole = 0;
4997         int ret;
4998
4999         if (argc != 1) {
5000                 usage("setrecmasterrole");
5001         }
5002
5003         if (strcmp(argv[0], "on") == 0) {
5004                 recmasterrole = 1;
5005         } else if (strcmp(argv[0], "off") == 0) {
5006                 recmasterrole = 0;
5007         } else {
5008                 usage("setrecmasterrole");
5009         }
5010
5011         ret = ctdb_ctrl_set_recmasterrole(mem_ctx, ctdb->ev, ctdb->client,
5012                                           ctdb->cmd_pnn, TIMEOUT(),
5013                                           recmasterrole);
5014         if (ret != 0) {
5015                 return ret;
5016         }
5017
5018         return 0;
5019 }
5020
5021 static int control_setdbreadonly(TALLOC_CTX *mem_ctx,
5022                                  struct ctdb_context *ctdb,
5023                                  int argc, const char **argv)
5024 {
5025         uint32_t db_id;
5026         uint8_t db_flags;
5027         int ret;
5028
5029         if (argc != 1) {
5030                 usage("setdbreadonly");
5031         }
5032
5033         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, NULL, &db_flags)) {
5034                 return 1;
5035         }
5036
5037         if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
5038                 fprintf(stderr, "Cannot set READONLY on persistent DB\n");
5039                 return 1;
5040         }
5041
5042         ret = ctdb_ctrl_set_db_readonly(mem_ctx, ctdb->ev, ctdb->client,
5043                                         ctdb->cmd_pnn, TIMEOUT(), db_id);
5044         if (ret != 0) {
5045                 return ret;
5046         }
5047
5048         return 0;
5049 }
5050
5051 static int control_setdbsticky(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5052                                int argc, const char **argv)
5053 {
5054         uint32_t db_id;
5055         uint8_t db_flags;
5056         int ret;
5057
5058         if (argc != 1) {
5059                 usage("setdbsticky");
5060         }
5061
5062         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, NULL, &db_flags)) {
5063                 return 1;
5064         }
5065
5066         if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
5067                 fprintf(stderr, "Cannot set STICKY on persistent DB\n");
5068                 return 1;
5069         }
5070
5071         ret = ctdb_ctrl_set_db_sticky(mem_ctx, ctdb->ev, ctdb->client,
5072                                       ctdb->cmd_pnn, TIMEOUT(), db_id);
5073         if (ret != 0) {
5074                 return ret;
5075         }
5076
5077         return 0;
5078 }
5079
5080 static int control_pfetch(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5081                           int argc, const char **argv)
5082 {
5083         const char *db_name;
5084         struct ctdb_db_context *db;
5085         struct ctdb_transaction_handle *h;
5086         uint8_t db_flags;
5087         TDB_DATA key, data;
5088         int ret;
5089
5090         if (argc < 2 || argc > 3) {
5091                 usage("pfetch");
5092         }
5093
5094         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5095                 return 1;
5096         }
5097
5098         if (! (db_flags & CTDB_DB_FLAGS_PERSISTENT)) {
5099                 fprintf(stderr, "DB %s is not a persistent database\n",
5100                         db_name);
5101                 return 1;
5102         }
5103
5104         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5105                           db_flags, &db);
5106         if (ret != 0) {
5107                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5108                 return ret;
5109         }
5110
5111         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5112         if (ret != 0) {
5113                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5114                 return ret;
5115         }
5116
5117         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
5118                                      TIMEOUT(), db, true, &h);
5119         if (ret != 0) {
5120                 fprintf(stderr, "Failed to start transaction on db %s\n",
5121                         db_name);
5122                 return ret;
5123         }
5124
5125         ret = ctdb_transaction_fetch_record(h, key, mem_ctx, &data);
5126         if (ret != 0) {
5127                 fprintf(stderr, "Failed to read record for key %s\n",
5128                         argv[1]);
5129                 return ret;
5130         }
5131
5132         printf("%.*s\n", (int)data.dsize, data.dptr);
5133
5134         ctdb_transaction_cancel(h);
5135         return 0;
5136 }
5137
5138 static int control_pstore(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5139                           int argc, const char **argv)
5140 {
5141         const char *db_name;
5142         struct ctdb_db_context *db;
5143         struct ctdb_transaction_handle *h;
5144         uint8_t db_flags;
5145         TDB_DATA key, data;
5146         int ret;
5147
5148         if (argc != 3) {
5149                 usage("pstore");
5150         }
5151
5152         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5153                 return 1;
5154         }
5155
5156         if (! (db_flags & CTDB_DB_FLAGS_PERSISTENT)) {
5157                 fprintf(stderr, "DB %s is not a persistent database\n",
5158                         db_name);
5159                 return 1;
5160         }
5161
5162         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5163                           db_flags, &db);
5164         if (ret != 0) {
5165                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5166                 return ret;
5167         }
5168
5169         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5170         if (ret != 0) {
5171                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5172                 return ret;
5173         }
5174
5175         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &data);
5176         if (ret != 0) {
5177                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
5178                 return ret;
5179         }
5180
5181         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
5182                                      TIMEOUT(), db, false, &h);
5183         if (ret != 0) {
5184                 fprintf(stderr, "Failed to start transaction on db %s\n",
5185                         db_name);
5186                 return ret;
5187         }
5188
5189         ret = ctdb_transaction_store_record(h, key, data);
5190         if (ret != 0) {
5191                 fprintf(stderr, "Failed to store record for key %s\n",
5192                         argv[1]);
5193                 return ret;
5194         }
5195
5196         ret = ctdb_transaction_commit(h);
5197         if (ret != 0) {
5198                 fprintf(stderr, "Failed to commit transaction on db %s\n",
5199                         db_name);
5200                 return ret;
5201         }
5202
5203         return 0;
5204 }
5205
5206 static int control_pdelete(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5207                            int argc, const char **argv)
5208 {
5209         const char *db_name;
5210         struct ctdb_db_context *db;
5211         struct ctdb_transaction_handle *h;
5212         uint8_t db_flags;
5213         TDB_DATA key;
5214         int ret;
5215
5216         if (argc != 2) {
5217                 usage("pdelete");
5218         }
5219
5220         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5221                 return 1;
5222         }
5223
5224         if (! (db_flags & CTDB_DB_FLAGS_PERSISTENT)) {
5225                 fprintf(stderr, "DB %s is not a persistent database\n",
5226                         db_name);
5227                 return 1;
5228         }
5229
5230         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5231                           db_flags, &db);
5232         if (ret != 0) {
5233                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5234                 return ret;
5235         }
5236
5237         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5238         if (ret != 0) {
5239                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5240                 return ret;
5241         }
5242
5243         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
5244                                      TIMEOUT(), db, false, &h);
5245         if (ret != 0) {
5246                 fprintf(stderr, "Failed to start transaction on db %s\n",
5247                         db_name);
5248                 return ret;
5249         }
5250
5251         ret = ctdb_transaction_delete_record(h, key);
5252         if (ret != 0) {
5253                 fprintf(stderr, "Failed to delete record for key %s\n",
5254                         argv[1]);
5255                 return ret;
5256         }
5257
5258         ret = ctdb_transaction_commit(h);
5259         if (ret != 0) {
5260                 fprintf(stderr, "Failed to commit transaction on db %s\n",
5261                         db_name);
5262                 return ret;
5263         }
5264
5265         return 0;
5266 }
5267
5268 static int ptrans_parse_string(TALLOC_CTX *mem_ctx, const char **ptr, TDB_DATA *data)
5269 {
5270         const char *t;
5271         size_t n;
5272         int ret;
5273
5274         *data = tdb_null;
5275
5276         /* Skip whitespace */
5277         n = strspn(*ptr, " \t");
5278         t = *ptr + n;
5279
5280         if (t[0] == '"') {
5281                 /* Quoted ASCII string - no wide characters! */
5282                 t++;
5283                 n = strcspn(t, "\"");
5284                 if (t[n] == '"') {
5285                         if (n > 0) {
5286                                 ret = str_to_data(t, n, mem_ctx, data);
5287                                 if (ret != 0) {
5288                                         return ret;
5289                                 }
5290                         }
5291                         *ptr = t + n + 1;
5292                 } else {
5293                         fprintf(stderr, "Unmatched \" in input %s\n", *ptr);
5294                         return 1;
5295                 }
5296         } else {
5297                 fprintf(stderr, "Unsupported input format in %s\n", *ptr);
5298                 return 1;
5299         }
5300
5301         return 0;
5302 }
5303
5304 #define MAX_LINE_SIZE   1024
5305
5306 static bool ptrans_get_key_value(TALLOC_CTX *mem_ctx, FILE *file,
5307                                  TDB_DATA *key, TDB_DATA *value)
5308 {
5309         char line [MAX_LINE_SIZE]; /* FIXME: make this more flexible? */
5310         const char *ptr;
5311         int ret;
5312
5313         ptr = fgets(line, MAX_LINE_SIZE, file);
5314         if (ptr == NULL) {
5315                 return false;
5316         }
5317
5318         /* Get key */
5319         ret = ptrans_parse_string(mem_ctx, &ptr, key);
5320         if (ret != 0 || ptr == NULL || key->dptr == NULL) {
5321                 /* Line Ignored but not EOF */
5322                 *key = tdb_null;
5323                 return true;
5324         }
5325
5326         /* Get value */
5327         ret = ptrans_parse_string(mem_ctx, &ptr, value);
5328         if (ret != 0) {
5329                 /* Line Ignored but not EOF */
5330                 talloc_free(key->dptr);
5331                 *key = tdb_null;
5332                 return true;
5333         }
5334
5335         return true;
5336 }
5337
5338 static int control_ptrans(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5339                           int argc, const char **argv)
5340 {
5341         const char *db_name;
5342         struct ctdb_db_context *db;
5343         struct ctdb_transaction_handle *h;
5344         uint8_t db_flags;
5345         FILE *file;
5346         TDB_DATA key, value;
5347         int ret;
5348
5349         if (argc < 1 || argc > 2) {
5350                 usage("ptrans");
5351         }
5352
5353         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5354                 return 1;
5355         }
5356
5357         if (! (db_flags & CTDB_DB_FLAGS_PERSISTENT)) {
5358                 fprintf(stderr, "DB %s is not a persistent database\n",
5359                         db_name);
5360                 return 1;
5361         }
5362
5363         if (argc == 2) {
5364                 file = fopen(argv[1], "r");
5365                 if (file == NULL) {
5366                         fprintf(stderr, "Failed to open file %s\n", argv[1]);
5367                         return 1;
5368                 }
5369         } else {
5370                 file = stdin;
5371         }
5372
5373         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5374                           db_flags, &db);
5375         if (ret != 0) {
5376                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5377                 goto done;
5378         }
5379
5380         ret = ctdb_transaction_start(mem_ctx, ctdb->ev, ctdb->client,
5381                                      TIMEOUT(), db, false, &h);
5382         if (ret != 0) {
5383                 fprintf(stderr, "Failed to start transaction on db %s\n",
5384                         db_name);
5385                 goto done;
5386         }
5387
5388         while (ptrans_get_key_value(mem_ctx, file, &key, &value)) {
5389                 if (key.dsize != 0) {
5390                         ret = ctdb_transaction_store_record(h, key, value);
5391                         if (ret != 0) {
5392                                 fprintf(stderr, "Failed to store record\n");
5393                                 ctdb_transaction_cancel(h);
5394                                 goto done;
5395                         }
5396                         talloc_free(key.dptr);
5397                         talloc_free(value.dptr);
5398                 }
5399         }
5400
5401         ret = ctdb_transaction_commit(h);
5402         if (ret != 0) {
5403                 fprintf(stderr, "Failed to commit transaction on db %s\n",
5404                         db_name);
5405         }
5406
5407 done:
5408         if (file != stdin) {
5409                 fclose(file);
5410         }
5411         return ret;
5412 }
5413
5414 static int control_tfetch(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5415                           int argc, const char **argv)
5416 {
5417         struct tdb_context *tdb;
5418         TDB_DATA key, data;
5419         struct ctdb_ltdb_header header;
5420         int ret;
5421
5422         if (argc < 2 || argc > 3) {
5423                 usage("tfetch");
5424         }
5425
5426         tdb = tdb_open(argv[0], 0, 0, O_RDWR, 0);
5427         if (tdb == NULL) {
5428                 fprintf(stderr, "Failed to open TDB file %s\n", argv[0]);
5429                 return 1;
5430         }
5431
5432         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5433         if (ret != 0) {
5434                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5435                 return ret;
5436         }
5437
5438         data = tdb_fetch(tdb, key);
5439         if (data.dptr == NULL) {
5440                 fprintf(stderr, "No record for key %s\n", argv[1]);
5441                 return 1;
5442         }
5443
5444         if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
5445                 fprintf(stderr, "Invalid record for key %s\n", argv[1]);
5446                 return 1;
5447         }
5448
5449         tdb_close(tdb);
5450
5451         if (argc == 3) {
5452                 int fd;
5453                 ssize_t nwritten;
5454
5455                 fd = open(argv[2], O_WRONLY|O_CREAT|O_TRUNC, 0600);
5456                 if (fd == -1) {
5457                         fprintf(stderr, "Failed to open output file %s\n",
5458                                 argv[2]);
5459                         goto fail;
5460                 }
5461
5462                 nwritten = sys_write(fd, data.dptr, data.dsize);
5463                 if (nwritten != data.dsize) {
5464                         fprintf(stderr, "Failed to write record to file\n");
5465                         close(fd);
5466                         goto fail;
5467                 }
5468
5469                 close(fd);
5470         }
5471
5472 fail:
5473         ret = ctdb_ltdb_header_extract(&data, &header);
5474         if (ret != 0) {
5475                 fprintf(stderr, "Failed to parse header from data\n");
5476                 return 1;
5477         }
5478
5479         dump_ltdb_header(&header);
5480         dump_tdb_data("data", data);
5481
5482         return 0;
5483 }
5484
5485 static int control_tstore(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5486                           int argc, const char **argv)
5487 {
5488         struct tdb_context *tdb;
5489         TDB_DATA key, data, value;
5490         struct ctdb_ltdb_header header;
5491         size_t offset;
5492         int ret;
5493
5494         if (argc < 3 || argc > 5) {
5495                 usage("tstore");
5496         }
5497
5498         tdb = tdb_open(argv[0], 0, 0, O_RDWR, 0);
5499         if (tdb == NULL) {
5500                 fprintf(stderr, "Failed to open TDB file %s\n", argv[0]);
5501                 return 1;
5502         }
5503
5504         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5505         if (ret != 0) {
5506                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5507                 return ret;
5508         }
5509
5510         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &value);
5511         if (ret != 0) {
5512                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
5513                 return ret;
5514         }
5515
5516         ZERO_STRUCT(header);
5517
5518         if (argc > 3) {
5519                 header.rsn = (uint64_t)strtoull(argv[3], NULL, 0);
5520         }
5521         if (argc > 4) {
5522                 header.dmaster = (uint32_t)atol(argv[4]);
5523         }
5524         if (argc > 5) {
5525                 header.flags = (uint32_t)atol(argv[5]);
5526         }
5527
5528         offset = ctdb_ltdb_header_len(&header);
5529         data.dsize = offset + value.dsize;
5530         data.dptr = talloc_size(mem_ctx, data.dsize);
5531         if (data.dptr == NULL) {
5532                 fprintf(stderr, "Memory allocation error\n");
5533                 return 1;
5534         }
5535
5536         ctdb_ltdb_header_push(&header, data.dptr);
5537         memcpy(data.dptr + offset, value.dptr, value.dsize);
5538
5539         ret = tdb_store(tdb, key, data, TDB_REPLACE);
5540         if (ret != 0) {
5541                 fprintf(stderr, "Failed to write record %s to file %s\n",
5542                         argv[1], argv[0]);
5543         }
5544
5545         tdb_close(tdb);
5546
5547         return ret;
5548 }
5549
5550 static int control_readkey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5551                            int argc, const char **argv)
5552 {
5553         const char *db_name;
5554         struct ctdb_db_context *db;
5555         struct ctdb_record_handle *h;
5556         uint8_t db_flags;
5557         TDB_DATA key, data;
5558         bool readonly = false;
5559         int ret;
5560
5561         if (argc < 2 || argc > 3) {
5562                 usage("readkey");
5563         }
5564
5565         if (argc == 3) {
5566                 if (strcmp(argv[2], "readonly") == 0) {
5567                         readonly = true;
5568                 } else {
5569                         usage("readkey");
5570                 }
5571         }
5572
5573         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5574                 return 1;
5575         }
5576
5577         if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
5578                 fprintf(stderr, "DB %s is not a volatile database\n",
5579                         db_name);
5580                 return 1;
5581         }
5582
5583         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5584                           db_flags, &db);
5585         if (ret != 0) {
5586                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5587                 return ret;
5588         }
5589
5590         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5591         if (ret != 0) {
5592                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5593                 return ret;
5594         }
5595
5596         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5597                               db, key, readonly, &h, NULL, &data);
5598         if (ret != 0) {
5599                 fprintf(stderr, "Failed to read record for key %s\n",
5600                         argv[1]);
5601         } else {
5602                 printf("Data: size:%zu ptr:[%.*s]\n", data.dsize,
5603                        (int)data.dsize, data.dptr);
5604         }
5605
5606         talloc_free(h);
5607         return ret;
5608 }
5609
5610 static int control_writekey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5611                             int argc, const char **argv)
5612 {
5613         const char *db_name;
5614         struct ctdb_db_context *db;
5615         struct ctdb_record_handle *h;
5616         uint8_t db_flags;
5617         TDB_DATA key, data;
5618         int ret;
5619
5620         if (argc != 3) {
5621                 usage("writekey");
5622         }
5623
5624         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5625                 return 1;
5626         }
5627
5628         if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
5629                 fprintf(stderr, "DB %s is not a volatile database\n",
5630                         db_name);
5631                 return 1;
5632         }
5633
5634         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5635                           db_flags, &db);
5636         if (ret != 0) {
5637                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5638                 return ret;
5639         }
5640
5641         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5642         if (ret != 0) {
5643                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5644                 return ret;
5645         }
5646
5647         ret = str_to_data(argv[2], strlen(argv[2]), mem_ctx, &data);
5648         if (ret != 0) {
5649                 fprintf(stderr, "Failed to parse value %s\n", argv[2]);
5650                 return ret;
5651         }
5652
5653         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5654                               db, key, false, &h, NULL, NULL);
5655         if (ret != 0) {
5656                 fprintf(stderr, "Failed to lock record for key %s\n", argv[0]);
5657                 return ret;
5658         }
5659
5660         ret = ctdb_store_record(h, data);
5661         if (ret != 0) {
5662                 fprintf(stderr, "Failed to store record for key %s\n",
5663                         argv[1]);
5664         }
5665
5666         talloc_free(h);
5667         return ret;
5668 }
5669
5670 static int control_deletekey(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5671                              int argc, const char **argv)
5672 {
5673         const char *db_name;
5674         struct ctdb_db_context *db;
5675         struct ctdb_record_handle *h;
5676         uint8_t db_flags;
5677         TDB_DATA key, data;
5678         int ret;
5679
5680         if (argc != 2) {
5681                 usage("deletekey");
5682         }
5683
5684         if (! db_exists(mem_ctx, ctdb, argv[0], NULL, &db_name, &db_flags)) {
5685                 return 1;
5686         }
5687
5688         if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
5689                 fprintf(stderr, "DB %s is not a volatile database\n",
5690                         db_name);
5691                 return 1;
5692         }
5693
5694         ret = ctdb_attach(ctdb->ev, ctdb->client, TIMEOUT(), db_name,
5695                           db_flags, &db);
5696         if (ret != 0) {
5697                 fprintf(stderr, "Failed to attach to DB %s\n", db_name);
5698                 return ret;
5699         }
5700
5701         ret = str_to_data(argv[1], strlen(argv[1]), mem_ctx, &key);
5702         if (ret != 0) {
5703                 fprintf(stderr, "Failed to parse key %s\n", argv[1]);
5704                 return ret;
5705         }
5706
5707         ret = ctdb_fetch_lock(mem_ctx, ctdb->ev, ctdb->client,
5708                               db, key, false, &h, NULL, &data);
5709         if (ret != 0) {
5710                 fprintf(stderr, "Failed to fetch record for key %s\n",
5711                         argv[1]);
5712                 return ret;
5713         }
5714
5715         ret = ctdb_delete_record(h);
5716         if (ret != 0) {
5717                 fprintf(stderr, "Failed to delete record for key %s\n",
5718                         argv[1]);
5719         }
5720
5721         talloc_free(h);
5722         return ret;
5723 }
5724
5725 static int control_checktcpport(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5726                                 int argc, const char **argv)
5727 {
5728         struct sockaddr_in sin;
5729         unsigned int port;
5730         int s, v;
5731         int ret;
5732
5733         if (argc != 1) {
5734                 usage("chktcpport");
5735         }
5736
5737         port = atoi(argv[0]);
5738
5739         s = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
5740         if (s == -1) {
5741                 fprintf(stderr, "Failed to open local socket\n");
5742                 return errno;
5743         }
5744
5745         v = fcntl(s, F_GETFL, 0);
5746         if (v == -1 || fcntl(s, F_SETFL, v | O_NONBLOCK)) {
5747                 fprintf(stderr, "Unable to set socket non-blocking\n");
5748                 close(s);
5749                 return errno;
5750         }
5751
5752         bzero(&sin, sizeof(sin));
5753         sin.sin_family = AF_INET;
5754         sin.sin_port = htons(port);
5755         ret = bind(s, (struct sockaddr *)&sin, sizeof(sin));
5756         close(s);
5757         if (ret == -1) {
5758                 fprintf(stderr, "Failed to bind to TCP port %u\n", port);
5759                 return errno;
5760         }
5761
5762         return 0;
5763 }
5764
5765 static int control_rebalancenode(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5766                                  int argc, const char **argv)
5767 {
5768         if (argc != 0) {
5769                 usage("rebalancenode");
5770         }
5771
5772         if (! rebalancenode(mem_ctx, ctdb, ctdb->cmd_pnn)) {
5773                 fprintf(stderr, "Failed to rebalance IPs on node %u\n",
5774                         ctdb->cmd_pnn);
5775                 return 1;
5776         }
5777
5778         return 0;
5779 }
5780
5781 static int control_getdbseqnum(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5782                                int argc, const char **argv)
5783 {
5784         uint32_t db_id;
5785         const char *db_name;
5786         uint64_t seqnum;
5787         int ret;
5788
5789         if (argc != 1) {
5790                 usage("getdbseqnum");
5791         }
5792
5793         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, NULL)) {
5794                 return 1;
5795         }
5796
5797         ret = ctdb_ctrl_get_db_seqnum(mem_ctx, ctdb->ev, ctdb->client,
5798                                       ctdb->cmd_pnn, TIMEOUT(), db_id,
5799                                       &seqnum);
5800         if (ret != 0) {
5801                 fprintf(stderr, "Failed to get sequence number for DB %s\n",
5802                         db_name);
5803                 return ret;
5804         }
5805
5806         printf("0x%"PRIx64"\n", seqnum);
5807         return 0;
5808 }
5809
5810 static int control_nodestatus(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5811                               int argc, const char **argv)
5812 {
5813         const char *nodestring = NULL;
5814         struct ctdb_node_map *nodemap;
5815         int ret, i;
5816
5817         if (argc > 1) {
5818                 usage("nodestatus");
5819         }
5820
5821         if (argc == 1) {
5822                 nodestring = argv[0];
5823         }
5824
5825         if (! parse_nodestring(mem_ctx, ctdb, nodestring, &nodemap)) {
5826                 return 1;
5827         }
5828
5829         nodemap = get_nodemap(ctdb, false);
5830         if (nodemap == NULL) {
5831                 return 1;
5832         }
5833
5834         if (options.machinereadable) {
5835                 print_nodemap_machine(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn);
5836         } else {
5837                 print_nodemap(mem_ctx, ctdb, nodemap, ctdb->cmd_pnn);
5838         }
5839
5840         ret = 0;
5841         for (i=0; i<nodemap->num; i++) {
5842                 ret |= nodemap->node[i].flags;
5843         }
5844
5845         return ret;
5846 }
5847
5848 const struct {
5849         const char *name;
5850         uint32_t offset;
5851 } db_stats_fields[] = {
5852 #define DBSTATISTICS_FIELD(n) { #n, offsetof(struct ctdb_db_statistics, n) }
5853         DBSTATISTICS_FIELD(db_ro_delegations),
5854         DBSTATISTICS_FIELD(db_ro_revokes),
5855         DBSTATISTICS_FIELD(locks.num_calls),
5856         DBSTATISTICS_FIELD(locks.num_current),
5857         DBSTATISTICS_FIELD(locks.num_pending),
5858         DBSTATISTICS_FIELD(locks.num_failed),
5859         DBSTATISTICS_FIELD(db_ro_delegations),
5860 };
5861
5862 static void print_dbstatistics(const char *db_name,
5863                                struct ctdb_db_statistics *s)
5864 {
5865         int i;
5866         const char *prefix = NULL;
5867         int preflen = 0;
5868
5869         printf("DB Statistics %s\n", db_name);
5870
5871         for (i=0; i<ARRAY_SIZE(db_stats_fields); i++) {
5872                 if (strchr(db_stats_fields[i].name, '.') != NULL) {
5873                         preflen = strcspn(db_stats_fields[i].name, ".") + 1;
5874                         if (! prefix ||
5875                             strncmp(prefix, db_stats_fields[i].name, preflen) != 0) {
5876                                 prefix = db_stats_fields[i].name;
5877                                 printf(" %*.*s\n", preflen-1, preflen-1,
5878                                        db_stats_fields[i].name);
5879                         }
5880                 } else {
5881                         preflen = 0;
5882                 }
5883                 printf(" %*s%-22s%*s%10u\n", preflen ? 4 : 0, "",
5884                        db_stats_fields[i].name+preflen, preflen ? 0 : 4, "",
5885                        *(uint32_t *)(db_stats_fields[i].offset+(uint8_t *)s));
5886         }
5887
5888         printf(" hop_count_buckets:");
5889         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
5890                 printf(" %d", s->hop_count_bucket[i]);
5891         }
5892         printf("\n");
5893
5894         printf(" lock_buckets:");
5895         for (i=0; i<MAX_COUNT_BUCKETS; i++) {
5896                 printf(" %d", s->locks.buckets[i]);
5897         }
5898         printf("\n");
5899
5900         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
5901                "locks_latency      MIN/AVG/MAX",
5902                s->locks.latency.min, LATENCY_AVG(s->locks.latency),
5903                s->locks.latency.max, s->locks.latency.num);
5904
5905         printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n",
5906                "vacuum_latency     MIN/AVG/MAX",
5907                s->vacuum.latency.min, LATENCY_AVG(s->vacuum.latency),
5908                s->vacuum.latency.max, s->vacuum.latency.num);
5909
5910         printf(" Num Hot Keys:     %d\n", s->num_hot_keys);
5911         for (i=0; i<s->num_hot_keys; i++) {
5912                 int j;
5913                 printf("     Count:%d Key:", s->hot_keys[i].count);
5914                 for (j=0; j<s->hot_keys[i].key.dsize; j++) {
5915                         printf("%02x", s->hot_keys[i].key.dptr[j] & 0xff);
5916                 }
5917                 printf("\n");
5918         }
5919 }
5920
5921 static int control_dbstatistics(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
5922                                 int argc, const char **argv)
5923 {
5924         uint32_t db_id;
5925         const char *db_name;
5926         struct ctdb_db_statistics *dbstats;
5927         int ret;
5928
5929         if (argc != 1) {
5930                 usage("dbstatistics");
5931         }
5932
5933         if (! db_exists(mem_ctx, ctdb, argv[0], &db_id, &db_name, NULL)) {
5934                 return 1;
5935         }
5936
5937         ret = ctdb_ctrl_get_db_statistics(mem_ctx, ctdb->ev, ctdb->client,
5938                                           ctdb->cmd_pnn, TIMEOUT(), db_id,
5939                                           &dbstats);
5940         if (ret != 0) {
5941                 fprintf(stderr, "Failed to get statistics for DB %s\n",
5942                         db_name);
5943                 return ret;
5944         }
5945
5946         print_dbstatistics(db_name, dbstats);
5947         return 0;
5948 }
5949
5950 struct disable_takeover_runs_state {
5951         uint32_t *pnn_list;
5952         int node_count;
5953         bool *reply;
5954         int status;
5955         bool done;
5956 };
5957
5958 static void disable_takeover_run_handler(uint64_t srvid, TDB_DATA data,
5959                                          void *private_data)
5960 {
5961         struct disable_takeover_runs_state *state =
5962                 (struct disable_takeover_runs_state *)private_data;
5963         int ret, i;
5964
5965         if (data.dsize != sizeof(int)) {
5966                 /* Ignore packet */
5967                 return;
5968         }
5969
5970         /* ret will be a PNN (i.e. >=0) on success, or negative on error */
5971         ret = *(int *)data.dptr;
5972         if (ret < 0) {
5973                 state->status = ret;
5974                 state->done = true;
5975                 return;
5976         }
5977         for (i=0; i<state->node_count; i++) {
5978                 if (state->pnn_list[i] == ret) {
5979                         state->reply[i] = true;
5980                         break;
5981                 }
5982         }
5983
5984         state->done = true;
5985         for (i=0; i<state->node_count; i++) {
5986                 if (! state->reply[i]) {
5987                         state->done = false;
5988                         break;
5989                 }
5990         }
5991 }
5992
5993 static int disable_takeover_runs(TALLOC_CTX *mem_ctx,
5994                                  struct ctdb_context *ctdb, uint32_t timeout,
5995                                  uint32_t *pnn_list, int count)
5996 {
5997         struct ctdb_disable_message disable = { 0 };
5998         struct disable_takeover_runs_state state;
5999         int ret, i;
6000
6001         disable.pnn = ctdb->pnn;
6002         disable.srvid = next_srvid(ctdb);
6003         disable.timeout = timeout;
6004
6005         state.pnn_list = pnn_list;
6006         state.node_count = count;
6007         state.done = false;
6008         state.status = 0;
6009         state.reply = talloc_zero_array(mem_ctx, bool, count);
6010         if (state.reply == NULL) {
6011                 return ENOMEM;
6012         }
6013
6014         ret = ctdb_client_set_message_handler(ctdb->ev, ctdb->client,
6015                                               disable.srvid,
6016                                               disable_takeover_run_handler,
6017                                               &state);
6018         if (ret != 0) {
6019                 return ret;
6020         }
6021
6022         for (i=0; i<count; i++) {
6023                 ret = ctdb_message_disable_takeover_runs(mem_ctx, ctdb->ev,
6024                                                          ctdb->client,
6025                                                          pnn_list[i],
6026                                                          &disable);
6027                 if (ret != 0) {
6028                         goto fail;
6029                 }
6030         }
6031
6032         ret = ctdb_client_wait_timeout(ctdb->ev, &state.done, TIMEOUT());
6033         if (ret == ETIME) {
6034                 fprintf(stderr, "Timed out waiting to disable takeover runs\n");
6035         } else {
6036                 ret = (state.status >= 0 ? 0 : 1);
6037         }
6038
6039 fail:
6040         ctdb_client_remove_message_handler(ctdb->ev, ctdb->client,
6041                                            disable.srvid, &state);
6042         return ret;
6043 }
6044
6045 static int control_reloadips(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
6046                              int argc, const char **argv)
6047 {
6048         const char *nodestring = NULL;
6049         struct ctdb_node_map *nodemap, *nodemap2;
6050         struct ctdb_req_control request;
6051         uint32_t *pnn_list, *pnn_list2;
6052         int ret, count, count2;
6053
6054         if (argc > 1) {
6055                 usage("reloadips");
6056         }
6057
6058         if (argc == 1) {
6059                 nodestring = argv[0];
6060         }
6061
6062         nodemap = get_nodemap(ctdb, false);
6063         if (nodemap == NULL) {
6064                 return 1;
6065         }
6066
6067         if (! parse_nodestring(mem_ctx, ctdb, nodestring, &nodemap2)) {
6068                 return 1;
6069         }
6070
6071         count = list_of_connected_nodes(nodemap, CTDB_UNKNOWN_PNN,
6072                                         mem_ctx, &pnn_list);
6073         if (count <= 0) {
6074                 fprintf(stderr, "Memory allocation error\n");
6075                 return 1;
6076         }
6077
6078         count2 = list_of_active_nodes(nodemap2, CTDB_UNKNOWN_PNN,
6079                                       mem_ctx, &pnn_list2);
6080         if (count2 <= 0) {
6081                 fprintf(stderr, "Memory allocation error\n");
6082                 return 1;
6083         }
6084
6085         /* Disable takeover runs on all connected nodes.  A reply
6086          * indicating success is needed from each node so all nodes
6087          * will need to be active.
6088          *
6089          * A check could be added to not allow reloading of IPs when
6090          * there are disconnected nodes.  However, this should
6091          * probably be left up to the administrator.
6092          */
6093         ret = disable_takeover_runs(mem_ctx, ctdb, 2*options.timelimit,
6094                                     pnn_list, count);
6095         if (ret != 0) {
6096                 fprintf(stderr, "Failed to disable takeover runs\n");
6097                 return ret;
6098         }
6099
6100         /* Now tell all the desired nodes to reload their public IPs.
6101          * Keep trying this until it succeeds.  This assumes all
6102          * failures are transient, which might not be true...
6103          */
6104         ctdb_req_control_reload_public_ips(&request);
6105         ret = ctdb_client_control_multi(mem_ctx, ctdb->ev, ctdb->client,
6106                                         pnn_list2, count2, TIMEOUT(),
6107                                         &request, NULL, NULL);
6108         if (ret != 0) {
6109                 fprintf(stderr, "Failed to reload IPs on some nodes.\n");
6110         }
6111
6112         /* It isn't strictly necessary to wait until takeover runs are
6113          * re-enabled but doing so can't hurt.
6114          */
6115         ret = disable_takeover_runs(mem_ctx, ctdb, 0, pnn_list, count);
6116         if (ret != 0) {
6117                 fprintf(stderr, "Failed to enable takeover runs\n");
6118                 return ret;
6119         }
6120
6121         return ipreallocate(mem_ctx, ctdb);
6122 }
6123
6124 static int control_ipiface(TALLOC_CTX *mem_ctx, struct ctdb_context *ctdb,
6125                            int argc, const char **argv)
6126 {
6127         ctdb_sock_addr addr;
6128         char *iface;
6129
6130         if (argc != 1) {
6131                 usage("ipiface");
6132         }
6133
6134         if (! parse_ip(argv[0], NULL, 0, &addr)) {
6135                 fprintf(stderr, "Failed to Parse IP %s\n", argv[0]);
6136                 return 1;
6137         }
6138
6139         iface = ctdb_sys_find_ifname(&addr);
6140         if (iface == NULL) {
6141                 fprintf(stderr, "Failed to find interface for IP %s\n",
6142                         argv[0]);
6143                 return 1;
6144         }
6145         free(iface);
6146
6147         return 0;
6148 }
6149
6150
6151 static const struct ctdb_cmd {
6152         const char *name;
6153         int (*fn)(TALLOC_CTX *, struct ctdb_context *, int, const char **);
6154         bool without_daemon; /* can be run without daemon running ? */
6155         bool remote; /* can be run on remote nodes */
6156         const char *msg;
6157         const char *args;
6158 } ctdb_commands[] = {
6159         { "version", control_version, true, false,
6160                 "show version of ctdb", NULL },
6161         { "status", control_status, false, true,
6162                 "show node status", NULL },
6163         { "uptime", control_uptime, false, true,
6164                 "show node uptime", NULL },
6165         { "ping", control_ping, false, true,
6166                 "ping all nodes", NULL },
6167         { "runstate", control_runstate, false, true,
6168                 "get/check runstate of a node",
6169                 "[setup|first_recovery|startup|running]" },
6170         { "getvar", control_getvar, false, true,
6171                 "get a tunable variable", "<name>" },
6172         { "setvar", control_setvar, false, true,
6173                 "set a tunable variable", "<name> <value>" },
6174         { "listvars", control_listvars, false, true,
6175                 "list tunable variables", NULL },
6176         { "statistics", control_statistics, false, true,
6177                 "show ctdb statistics", NULL },
6178         { "statisticsreset", control_statistics_reset, false, true,
6179                 "reset ctdb statistics", NULL },
6180         { "stats", control_stats, false, true,
6181                 "show rolling statistics", "[count]" },
6182         { "ip", control_ip, false, true,
6183                 "show public ips", "[all]" },
6184         { "ipinfo", control_ipinfo, false, true,
6185                 "show public ip details", "<ip>" },
6186         { "ifaces", control_ifaces, false, true,
6187                 "show interfaces", NULL },
6188         { "setifacelink", control_setifacelink, false, true,
6189                 "set interface link status", "<iface> up|down" },
6190         { "process-exists", control_process_exists, false, true,
6191                 "check if a process exists on a node",  "<pid>" },
6192         { "getdbmap", control_getdbmap, false, true,
6193                 "show attached databases", NULL },
6194         { "getdbstatus", control_getdbstatus, false, true,
6195                 "show database status", "<dbname|dbid>" },
6196         { "catdb", control_catdb, false, false,
6197                 "dump cluster-wide ctdb database", "<dbname|dbid>" },
6198         { "cattdb", control_cattdb, false, false,
6199                 "dump local ctdb database", "<dbname|dbid>" },
6200         { "getmonmode", control_getmonmode, false, true,
6201                 "show monitoring mode", NULL },
6202         { "getcapabilities", control_getcapabilities, false, true,
6203                 "show node capabilities", NULL },
6204         { "pnn", control_pnn, false, false,
6205                 "show the pnn of the currnet node", NULL },
6206         { "lvs", control_lvs, false, false,
6207                 "show lvs configuration", "master|list|status" },
6208         { "disablemonitor", control_disable_monitor, false, true,
6209                 "disable monitoring", NULL },
6210         { "enablemonitor", control_enable_monitor, false, true,
6211                 "enable monitoring", NULL },
6212         { "setdebug", control_setdebug, false, true,
6213                 "set debug level", "ERROR|WARNING|NOTICE|INFO|DEBUG" },
6214         { "getdebug", control_getdebug, false, true,
6215                 "get debug level", NULL },
6216         { "attach", control_attach, false, false,
6217                 "attach a database", "<dbname> [persistent]" },
6218         { "detach", control_detach, false, false,
6219                 "detach database(s)", "<dbname|dbid> ..." },
6220         { "dumpmemory", control_dumpmemory, false, true,
6221                 "dump ctdbd memory map", NULL },
6222         { "rddumpmemory", control_rddumpmemory, false, true,
6223                 "dump recoverd memory map", NULL },
6224         { "getpid", control_getpid, false, true,
6225                 "get ctdbd process ID", NULL },
6226         { "disable", control_disable, false, true,
6227                 "disable a node", NULL },
6228         { "enable", control_enable, false, true,
6229                 "enable a node", NULL },
6230         { "stop", control_stop, false, true,
6231                 "stop a node", NULL },
6232         { "continue", control_continue, false, true,
6233                 "continue a stopped node", NULL },
6234         { "ban", control_ban, false, true,
6235                 "ban a node", "<bantime>"},
6236         { "unban", control_unban, false, true,
6237                 "unban a node", NULL },
6238         { "shutdown", control_shutdown, false, true,
6239                 "shutdown ctdb daemon", NULL },
6240         { "recover", control_recover, false, true,
6241                 "force recovery", NULL },
6242         { "sync", control_ipreallocate, false, true,
6243                 "run ip reallocation (deprecated)", NULL },
6244         { "ipreallocate", control_ipreallocate, false, true,
6245                 "run ip reallocation", NULL },
6246         { "isnotrecmaster", control_isnotrecmaster, false, false,
6247                 "check if local node is the recmaster", NULL },
6248         { "gratarp", control_gratarp, false, true,
6249                 "send a gratuitous arp", "<ip> <interface>" },
6250         { "tickle", control_tickle, false, false,
6251                 "send a tcp tickle ack", "<srcip:port> <dstip:port>" },
6252         { "gettickles", control_gettickles, false, true,
6253                 "get the list of tickles", "<ip> [<port>]" },
6254         { "addtickle", control_addtickle, false, true,
6255                 "add a tickle", "<ip>:<port> <ip>:<port>" },
6256         { "deltickle", control_deltickle, false, true,
6257                 "delete a tickle", "<ip>:<port> <ip>:<port>" },
6258         { "check_srvids", control_check_srvids, false, true,
6259                 "check if srvid is registered", "<id> [<id> ...]" },
6260         { "listnodes", control_listnodes, true, true,
6261                 "list nodes in the cluster", NULL },
6262         { "reloadnodes", control_reloadnodes, false, false,
6263                 "reload the nodes file all nodes", NULL },
6264         { "moveip", control_moveip, false, false,
6265                 "move an ip address to another node", "<ip> <node>" },
6266         { "rebalanceip", control_rebalanceip, false, false,
6267                 "move an ip address optimally to another node", "<ip>" },
6268         { "addip", control_addip, false, true,
6269                 "add an ip address to a node", "<ip/mask> <iface>" },
6270         { "delip", control_delip, false, true,
6271                 "delete an ip address from a node", "<ip>" },
6272         { "eventscript", control_eventscript, false, true,
6273                 "run an event", "monitor" },
6274         { "backupdb", control_backupdb, false, false,
6275                 "backup a database into a file", "<dbname|dbid> <file>" },
6276         { "restoredb", control_restoredb, false, false,
6277                 "restore a database from a file", "<file> [dbname]" },
6278         { "dumpdbbackup", control_dumpdbbackup, true, false,
6279                 "dump database from a backup file", "<file>" },
6280         { "wipedb", control_wipedb, false, false,
6281                 "wipe the contents of a database.", "<dbname|dbid>"},
6282         { "recmaster", control_recmaster, false, true,
6283                 "show the pnn for the recovery master", NULL },
6284         { "scriptstatus", control_scriptstatus, false, true,
6285                 "show event script status",
6286                 "[init|setup|startup|monitor|takeip|releaseip|ipreallocated]" },
6287         { "enablescript", control_enablescript, false, true,
6288                 "enable an eventscript", "<script>"},
6289         { "disablescript", control_disablescript, false, true,
6290                 "disable an eventscript", "<script>"},
6291         { "natgw", control_natgw, false, false,
6292                 "show natgw configuration", "master|list|status" },
6293         { "natgwlist", control_natgwlist, false, false,
6294                 "show the nodes belonging to this natgw configuration", NULL },
6295         { "getreclock", control_getreclock, false, true,
6296                 "get recovery lock file", NULL },
6297         { "setlmasterrole", control_setlmasterrole, false, true,
6298                 "set LMASTER role", "on|off" },
6299         { "setrecmasterrole", control_setrecmasterrole, false, true,
6300                 "set RECMASTER role", "on|off"},
6301         { "setdbreadonly", control_setdbreadonly, false, true,
6302                 "enable readonly records", "<dbname|dbid>" },
6303         { "setdbsticky", control_setdbsticky, false, true,
6304                 "enable sticky records", "<dbname|dbid>"},
6305         { "pfetch", control_pfetch, false, false,
6306                 "fetch record from persistent database", "<dbname|dbid> <key> [<file>]" },
6307         { "pstore", control_pstore, false, false,
6308                 "write record to persistent database", "<dbname|dbid> <key> <value>" },
6309         { "pdelete", control_pdelete, false, false,
6310                 "delete record from persistent database", "<dbname|dbid> <key>" },
6311         { "ptrans", control_ptrans, false, false,
6312                 "update a persistent database (from file or stdin)", "<dbname|dbid> [<file>]" },
6313         { "tfetch", control_tfetch, false, true,
6314                 "fetch a record", "<tdb-file> <key> [<file>]" },
6315         { "tstore", control_tstore, false, true,
6316                 "store a record", "<tdb-file> <key> <data> [<rsn> <dmaster> <flags>]" },
6317         { "readkey", control_readkey, false, false,
6318                 "read value of a database key", "<dbname|dbid> <key> [readonly]" },
6319         { "writekey", control_writekey, false, false,
6320                 "write value for a database key", "<dbname|dbid> <key> <value>" },
6321         { "deletekey", control_deletekey, false, false,
6322                 "delete a database key", "<dbname|dbid> <key>" },
6323         { "checktcpport", control_checktcpport, true, false,
6324                 "check if a service is bound to a specific tcp port or not", "<port>" },
6325         { "rebalancenode", control_rebalancenode, false, true,
6326                 "mark nodes as forced IP rebalancing targets", NULL },
6327         { "getdbseqnum", control_getdbseqnum, false, false,
6328                 "get database sequence number", "<dbname|dbid>" },
6329         { "nodestatus", control_nodestatus, false, true,
6330                 "show and return node status", "[all|<pnn-list>]" },
6331         { "dbstatistics", control_dbstatistics, false, true,
6332                 "show database statistics", "<dbname|dbid>" },
6333         { "reloadips", control_reloadips, false, false,
6334                 "reload the public addresses file", "[all|<pnn-list>]" },
6335         { "ipiface", control_ipiface, true, false,
6336                 "Find the interface an ip address is hosted on", "<ip>" },
6337 };
6338
6339 static const struct ctdb_cmd *match_command(const char *command)
6340 {
6341         const struct ctdb_cmd *cmd;
6342         int i;
6343
6344         for (i=0; i<ARRAY_SIZE(ctdb_commands); i++) {
6345                 cmd = &ctdb_commands[i];
6346                 if (strlen(command) == strlen(cmd->name) &&
6347                     strncmp(command, cmd->name, strlen(command)) == 0) {
6348                         return cmd;
6349                 }
6350         }
6351
6352         return NULL;
6353 }
6354
6355
6356 /**
6357  * Show usage message
6358  */
6359 static void usage_full(void)
6360 {
6361         int i;
6362
6363         poptPrintHelp(pc, stdout, 0);
6364         printf("\nCommands:\n");
6365         for (i=0; i<ARRAY_SIZE(ctdb_commands); i++) {
6366                 printf("  %-15s %-27s  %s\n",
6367                        ctdb_commands[i].name,
6368                        ctdb_commands[i].args ? ctdb_commands[i].args : "",
6369                        ctdb_commands[i].msg);
6370         }
6371 }
6372
6373 static void usage(const char *command)
6374 {
6375         const struct ctdb_cmd *cmd;
6376
6377         if (command == NULL) {
6378                 usage_full();
6379                 exit(1);
6380         }
6381
6382         cmd = match_command(command);
6383         if (cmd == NULL) {
6384                 usage_full();
6385         } else {
6386                 poptPrintUsage(pc, stdout, 0);
6387                 printf("\nCommands:\n");
6388                 printf("  %-15s %-27s  %s\n",
6389                        cmd->name, cmd->args ? cmd->args : "", cmd->msg);
6390         }
6391
6392         exit(1);
6393 }
6394
6395 struct poptOption cmdline_options[] = {
6396         POPT_AUTOHELP
6397         { "socket", 's', POPT_ARG_STRING, &options.socket, 0,
6398                 "CTDB socket path", "filename" },
6399         { "debug", 'd', POPT_ARG_STRING, &options.debuglevelstr, 0,
6400                 "debug level"},
6401         { "timelimit", 't', POPT_ARG_INT, &options.timelimit, 0,
6402                 "timelimit (in seconds)" },
6403         { "node", 'n', POPT_ARG_INT, &options.pnn, 0,
6404                 "node specification - integer" },
6405         { NULL, 'Y', POPT_ARG_NONE, &options.machinereadable, 0,
6406                 "enable machine readable output", NULL },
6407         { "separator", 'x', POPT_ARG_STRING, &options.sep, 0,
6408                 "specify separator for machine readable output", "CHAR" },
6409         { NULL, 'X', POPT_ARG_NONE, &options.machineparsable, 0,
6410                 "enable machine parsable output with separator |", NULL },
6411         { "verbose", 'v', POPT_ARG_NONE, &options.verbose, 0,
6412                 "enable verbose output", NULL },
6413         { "maxruntime", 'T', POPT_ARG_INT, &options.maxruntime, 0,
6414                 "die if runtime exceeds this limit (in seconds)" },
6415         POPT_TABLEEND
6416 };
6417
6418 static int process_command(const struct ctdb_cmd *cmd, int argc,
6419                            const char **argv)
6420 {
6421         TALLOC_CTX *tmp_ctx;
6422         struct ctdb_context *ctdb;
6423         int ret;
6424         bool status;
6425         uint64_t srvid_offset;
6426
6427         tmp_ctx = talloc_new(NULL);
6428         if (tmp_ctx == NULL) {
6429                 fprintf(stderr, "Memory allocation error\n");
6430                 goto fail;
6431         }
6432
6433         if (cmd->without_daemon) {
6434                 if (options.pnn != -1) {
6435                         fprintf(stderr,
6436                                 "Cannot specify node for command %s\n",
6437                                 cmd->name);
6438                         goto fail;
6439                 }
6440
6441                 return cmd->fn(tmp_ctx, NULL, argc-1, argv+1);
6442         }
6443
6444         ctdb = talloc_zero(tmp_ctx, struct ctdb_context);
6445         if (ctdb == NULL) {
6446                 fprintf(stderr, "Memory allocation error\n");
6447                 goto fail;
6448         }
6449
6450         ctdb->ev = tevent_context_init(ctdb);
6451         if (ctdb->ev == NULL) {
6452                 fprintf(stderr, "Failed to initialize tevent\n");
6453                 goto fail;
6454         }
6455
6456         ret = ctdb_client_init(ctdb, ctdb->ev, options.socket, &ctdb->client);
6457         if (ret != 0) {
6458                 fprintf(stderr, "Failed to connect to CTDB daemon (%s)\n",
6459                         options.socket);
6460
6461                 if (!find_node_xpnn(ctdb, NULL)) {
6462                         fprintf(stderr, "Is this node part of CTDB cluster?\n");
6463                 }
6464                 goto fail;
6465         }
6466
6467         ctdb->pnn = ctdb_client_pnn(ctdb->client);
6468         srvid_offset = getpid() & 0xFFFF;
6469         ctdb->srvid = SRVID_CTDB_TOOL | (srvid_offset << 16);
6470
6471         if (options.pnn != -1) {
6472                 status = verify_pnn(ctdb, options.pnn);
6473                 if (! status) {
6474                         goto fail;
6475                 }
6476
6477                 ctdb->cmd_pnn = options.pnn;
6478         } else {
6479                 ctdb->cmd_pnn = ctdb->pnn;
6480         }
6481
6482         if (! cmd->remote && ctdb->pnn != ctdb->cmd_pnn) {
6483                 fprintf(stderr, "Node cannot be specified for command %s\n",
6484                         cmd->name);
6485                 goto fail;
6486         }
6487
6488         ret = cmd->fn(tmp_ctx, ctdb, argc-1, argv+1);
6489         talloc_free(tmp_ctx);
6490         return ret;
6491
6492 fail:
6493         talloc_free(tmp_ctx);
6494         return 1;
6495 }
6496
6497 static void signal_handler(int sig)
6498 {
6499         fprintf(stderr, "Maximum runtime exceeded - exiting\n");
6500 }
6501
6502 static void alarm_handler(int sig)
6503 {
6504         /* Kill any child processes */
6505         signal(SIGTERM, signal_handler);
6506         kill(0, SIGTERM);
6507
6508         _exit(1);
6509 }
6510
6511 int main(int argc, const char *argv[])
6512 {
6513         int opt;
6514         const char **extra_argv;
6515         int extra_argc;
6516         const struct ctdb_cmd *cmd;
6517         const char *ctdb_socket;
6518         enum debug_level loglevel;
6519         int ret;
6520
6521         setlinebuf(stdout);
6522
6523         /* Set default options */
6524         options.socket = CTDB_SOCKET;
6525         options.debuglevelstr = NULL;
6526         options.timelimit = 10;
6527         options.sep = "|";
6528         options.maxruntime = 0;
6529         options.pnn = -1;
6530
6531         ctdb_socket = getenv("CTDB_SOCKET");
6532         if (ctdb_socket != NULL) {
6533                 options.socket = ctdb_socket;
6534         }
6535
6536         pc = poptGetContext(argv[0], argc, argv, cmdline_options,
6537                             POPT_CONTEXT_KEEP_FIRST);
6538         while ((opt = poptGetNextOpt(pc)) != -1) {
6539                 fprintf(stderr, "Invalid option %s: %s\n",
6540                         poptBadOption(pc, 0), poptStrerror(opt));
6541                 exit(1);
6542         }
6543
6544         if (options.maxruntime == 0) {
6545                 const char *ctdb_timeout;
6546
6547                 ctdb_timeout = getenv("CTDB_TIMEOUT");
6548                 if (ctdb_timeout != NULL) {
6549                         options.maxruntime = strtoul(ctdb_timeout, NULL, 0);
6550                 } else {
6551                         options.maxruntime = 120;
6552                 }
6553         }
6554         if (options.maxruntime <= 120) {
6555                 /* default timeout is 120 seconds */
6556                 options.maxruntime = 120;
6557         }
6558
6559         if (options.machineparsable) {
6560                 options.machinereadable = 1;
6561         }
6562
6563         /* setup the remaining options for the commands */
6564         extra_argc = 0;
6565         extra_argv = poptGetArgs(pc);
6566         if (extra_argv) {
6567                 extra_argv++;
6568                 while (extra_argv[extra_argc]) extra_argc++;
6569         }
6570
6571         if (extra_argc < 1) {
6572                 usage(NULL);
6573         }
6574
6575         cmd = match_command(extra_argv[0]);
6576         if (cmd == NULL) {
6577                 fprintf(stderr, "Unknown command '%s'\n", extra_argv[0]);
6578                 exit(1);
6579         }
6580
6581         /* Enable logging */
6582         setup_logging("ctdb", DEBUG_STDERR);
6583         if (debug_level_parse(options.debuglevelstr, &loglevel)) {
6584                 DEBUGLEVEL = loglevel;
6585         } else {
6586                 DEBUGLEVEL = DEBUG_ERR;
6587         }
6588
6589         signal(SIGALRM, alarm_handler);
6590         alarm(options.maxruntime);
6591
6592         ret = process_command(cmd, extra_argc, extra_argv);
6593         if (ret == -1) {
6594                 ret = 1;
6595         }
6596
6597         (void)poptFreeContext(pc);
6598
6599         return ret;
6600 }