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