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