tools/ctdb.c

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