ctdb/tools/ctdb.c

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