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