ctdb/tools/ctdb.c

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