ctdb/tools/ctdb.c

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