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