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