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 "lib/tevent/tevent.h"
  23 #include "system/time.h"
  24 #include "system/filesys.h"
  25 #include "system/network.h"
  26 #include "system/locale.h"
  27 #include "popt.h"
  28 #include "cmdline.h"
  29 #include "../include/ctdb.h"
  30 #include "../include/ctdb_client.h"
  31 #include "../include/ctdb_private.h"
  32 #include "../common/rb_tree.h"
  33 #include "db_wrap.h"
  34
  35 #define ERR_TIMEOUT     20      /* timed out trying to reach node */
  36 #define ERR_NONODE      21      /* node does not exist */
  37 #define ERR_DISNODE     22      /* node is disconnected */
  38
  39 struct ctdb_connection *ctdb_connection;
  40
  41 static void usage(void);
  42
  43 static struct {
  44         int timelimit;
  45         uint32_t pnn;
  46         int machinereadable;
  47         int verbose;
  48         int maxruntime;
  49 } options;
  50
  51 #define TIMELIMIT() timeval_current_ofs(options.timelimit, 0)
  52 #define LONGTIMELIMIT() timeval_current_ofs(options.timelimit*10, 0)
  53
  54 #ifdef CTDB_VERS
  55 static int control_version(struct ctdb_context *ctdb, int argc, const char **argv)
  56 {
  57 #define STR(x) #x
  58 #define XSTR(x) STR(x)
  59         printf("CTDB version: %s\n", XSTR(CTDB_VERS));
  60         return 0;
  61 }
  62 #endif
  63
  64
  65 /*
  66   verify that a node exists and is reachable
  67  */
  68 static void verify_node(struct ctdb_context *ctdb)
  69 {
  70         int ret;
  71         struct ctdb_node_map *nodemap=NULL;
  72
  73         if (options.pnn == CTDB_CURRENT_NODE) {
  74                 return;
  75         }
  76         if (options.pnn == CTDB_BROADCAST_ALL) {
  77                 return;
  78         }
  79
  80         /* verify the node exists */
  81         if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
  82                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
  83                 exit(10);
  84         }
  85         if (options.pnn >= nodemap->num) {
  86                 DEBUG(DEBUG_ERR, ("Node %u does not exist\n", options.pnn));
  87                 exit(ERR_NONODE);
  88         }
  89         if (nodemap->nodes[options.pnn].flags & NODE_FLAGS_DELETED) {
  90                 DEBUG(DEBUG_ERR, ("Node %u is DELETED\n", options.pnn));
  91                 exit(ERR_DISNODE);
  92         }
  93         if (nodemap->nodes[options.pnn].flags & NODE_FLAGS_DISCONNECTED) {
  94                 DEBUG(DEBUG_ERR, ("Node %u is DISCONNECTED\n", options.pnn));
  95                 exit(ERR_DISNODE);
  96         }
  97
  98         /* verify we can access the node */
  99         ret = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);
 100         if (ret == -1) {
 101                 DEBUG(DEBUG_ERR,("Can not access node. Node is not operational.\n"));
 102                 exit(10);
 103         }
 104 }
 105
 106 /*
 107  check if a database exists
 108 */
 109 static int db_exists(struct ctdb_context *ctdb, const char *db_name)
 110 {
 111         int i, ret;
 112         struct ctdb_dbid_map *dbmap=NULL;
 113
 114         ret = ctdb_ctrl_getdbmap(ctdb, TIMELIMIT(), options.pnn, ctdb, &dbmap);
 115         if (ret != 0) {
 116                 DEBUG(DEBUG_ERR, ("Unable to get dbids from node %u\n", options.pnn));
 117                 return -1;
 118         }
 119
 120         for(i=0;i<dbmap->num;i++){
 121                 const char *name;
 122
 123                 ctdb_ctrl_getdbname(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, ctdb, &name);
 124                 if (!strcmp(name, db_name)) {
 125                         return 0;
 126                 }
 127         }
 128
 129         return -1;
 130 }
 131
 132 /*
 133   see if a process exists
 134  */
 135 static int control_process_exists(struct ctdb_context *ctdb, int argc, const char **argv)
 136 {
 137         uint32_t pnn, pid;
 138         int ret;
 139         if (argc < 1) {
 140                 usage();
 141         }
 142
 143         if (sscanf(argv[0], "%u:%u", &pnn, &pid) != 2) {
 144                 DEBUG(DEBUG_ERR, ("Badly formed pnn:pid\n"));
 145                 return -1;
 146         }
 147
 148         ret = ctdb_ctrl_process_exists(ctdb, pnn, pid);
 149         if (ret == 0) {
 150                 printf("%u:%u exists\n", pnn, pid);
 151         } else {
 152                 printf("%u:%u does not exist\n", pnn, pid);
 153         }
 154         return ret;
 155 }
 156
 157 /*
 158   display statistics structure
 159  */
 160 static void show_statistics(struct ctdb_statistics *s)
 161 {
 162         TALLOC_CTX *tmp_ctx = talloc_new(NULL);
 163         int i;
 164         const char *prefix=NULL;
 165         int preflen=0;
 166         int tmp, days, hours, minutes, seconds;
 167         const struct {
 168                 const char *name;
 169                 uint32_t offset;
 170         } fields[] = {
 171 #define STATISTICS_FIELD(n) { #n, offsetof(struct ctdb_statistics, n) }
 172                 STATISTICS_FIELD(num_clients),
 173                 STATISTICS_FIELD(frozen),
 174                 STATISTICS_FIELD(recovering),
 175                 STATISTICS_FIELD(num_recoveries),
 176                 STATISTICS_FIELD(client_packets_sent),
 177                 STATISTICS_FIELD(client_packets_recv),
 178                 STATISTICS_FIELD(node_packets_sent),
 179                 STATISTICS_FIELD(node_packets_recv),
 180                 STATISTICS_FIELD(keepalive_packets_sent),
 181                 STATISTICS_FIELD(keepalive_packets_recv),
 182                 STATISTICS_FIELD(node.req_call),
 183                 STATISTICS_FIELD(node.reply_call),
 184                 STATISTICS_FIELD(node.req_dmaster),
 185                 STATISTICS_FIELD(node.reply_dmaster),
 186                 STATISTICS_FIELD(node.reply_error),
 187                 STATISTICS_FIELD(node.req_message),
 188                 STATISTICS_FIELD(node.req_control),
 189                 STATISTICS_FIELD(node.reply_control),
 190                 STATISTICS_FIELD(client.req_call),
 191                 STATISTICS_FIELD(client.req_message),
 192                 STATISTICS_FIELD(client.req_control),
 193                 STATISTICS_FIELD(timeouts.call),
 194                 STATISTICS_FIELD(timeouts.control),
 195                 STATISTICS_FIELD(timeouts.traverse),
 196                 STATISTICS_FIELD(total_calls),
 197                 STATISTICS_FIELD(pending_calls),
 198                 STATISTICS_FIELD(lockwait_calls),
 199                 STATISTICS_FIELD(pending_lockwait_calls),
 200                 STATISTICS_FIELD(childwrite_calls),
 201                 STATISTICS_FIELD(pending_childwrite_calls),
 202                 STATISTICS_FIELD(memory_used),
 203                 STATISTICS_FIELD(max_hop_count),
 204         };
 205         tmp = s->statistics_current_time.tv_sec - s->statistics_start_time.tv_sec;
 206         seconds = tmp%60;
 207         tmp    /= 60;
 208         minutes = tmp%60;
 209         tmp    /= 60;
 210         hours   = tmp%24;
 211         tmp    /= 24;
 212         days    = tmp;
 213
 214         printf("CTDB version %u\n", CTDB_VERSION);
 215         printf("Current time of statistics  :                %s", ctime(&s->statistics_current_time.tv_sec));
 216         printf("Statistics collected since  : (%03d %02d:%02d:%02d) %s", days, hours, minutes, seconds, ctime(&s->statistics_start_time.tv_sec));
 217
 218         for (i=0;i<ARRAY_SIZE(fields);i++) {
 219                 if (strchr(fields[i].name, '.')) {
 220                         preflen = strcspn(fields[i].name, ".")+1;
 221                         if (!prefix || strncmp(prefix, fields[i].name, preflen) != 0) {
 222                                 prefix = fields[i].name;
 223                                 printf(" %*.*s\n", preflen-1, preflen-1, fields[i].name);
 224                         }
 225                 } else {
 226                         preflen = 0;
 227                 }
 228                 printf(" %*s%-22s%*s%10u\n",
 229                        preflen?4:0, "",
 230                        fields[i].name+preflen,
 231                        preflen?0:4, "",
 232                        *(uint32_t *)(fields[i].offset+(uint8_t *)s));
 233         }
 234         printf(" %-30s     %.6f sec\n", "max_reclock_ctdbd", s->reclock.ctdbd);
 235         printf(" %-30s     %.6f sec\n", "max_reclock_recd", s->reclock.recd);
 236
 237         printf(" %-30s     %.6f sec\n", "max_call_latency", s->max_call_latency);
 238         printf(" %-30s     %.6f sec\n", "max_lockwait_latency", s->max_lockwait_latency);
 239         printf(" %-30s     %.6f sec\n", "max_childwrite_latency", s->max_childwrite_latency);
 240         printf(" %-30s     %.6f sec\n", "max_childwrite_latency", s->max_childwrite_latency);
 241
 242         talloc_free(tmp_ctx);
 243 }
 244
 245 /*
 246   display remote ctdb statistics combined from all nodes
 247  */
 248 static int control_statistics_all(struct ctdb_context *ctdb)
 249 {
 250         int ret, i;
 251         struct ctdb_statistics statistics;
 252         uint32_t *nodes;
 253         uint32_t num_nodes;
 254
 255         nodes = ctdb_get_connected_nodes(ctdb, TIMELIMIT(), ctdb, &num_nodes);
 256         CTDB_NO_MEMORY(ctdb, nodes);
 257
 258         ZERO_STRUCT(statistics);
 259
 260         for (i=0;i<num_nodes;i++) {
 261                 struct ctdb_statistics s1;
 262                 int j;
 263                 uint32_t *v1 = (uint32_t *)&s1;
 264                 uint32_t *v2 = (uint32_t *)&statistics;
 265                 uint32_t num_ints =
 266                         offsetof(struct ctdb_statistics, __last_counter) / sizeof(uint32_t);
 267                 ret = ctdb_ctrl_statistics(ctdb, nodes[i], &s1);
 268                 if (ret != 0) {
 269                         DEBUG(DEBUG_ERR, ("Unable to get statistics from node %u\n", nodes[i]));
 270                         return ret;
 271                 }
 272                 for (j=0;j<num_ints;j++) {
 273                         v2[j] += v1[j];
 274                 }
 275                 statistics.max_hop_count =
 276                         MAX(statistics.max_hop_count, s1.max_hop_count);
 277                 statistics.max_call_latency =
 278                         MAX(statistics.max_call_latency, s1.max_call_latency);
 279                 statistics.max_lockwait_latency =
 280                         MAX(statistics.max_lockwait_latency, s1.max_lockwait_latency);
 281         }
 282         talloc_free(nodes);
 283         printf("Gathered statistics for %u nodes\n", num_nodes);
 284         show_statistics(&statistics);
 285         return 0;
 286 }
 287
 288 /*
 289   display remote ctdb statistics
 290  */
 291 static int control_statistics(struct ctdb_context *ctdb, int argc, const char **argv)
 292 {
 293         int ret;
 294         struct ctdb_statistics statistics;
 295
 296         if (options.pnn == CTDB_BROADCAST_ALL) {
 297                 return control_statistics_all(ctdb);
 298         }
 299
 300         ret = ctdb_ctrl_statistics(ctdb, options.pnn, &statistics);
 301         if (ret != 0) {
 302                 DEBUG(DEBUG_ERR, ("Unable to get statistics from node %u\n", options.pnn));
 303                 return ret;
 304         }
 305         show_statistics(&statistics);
 306         return 0;
 307 }
 308
 309
 310 /*
 311   reset remote ctdb statistics
 312  */
 313 static int control_statistics_reset(struct ctdb_context *ctdb, int argc, const char **argv)
 314 {
 315         int ret;
 316
 317         ret = ctdb_statistics_reset(ctdb, options.pnn);
 318         if (ret != 0) {
 319                 DEBUG(DEBUG_ERR, ("Unable to reset statistics on node %u\n", options.pnn));
 320                 return ret;
 321         }
 322         return 0;
 323 }
 324
 325
 326 /*
 327   display remote ctdb rolling statistics
 328  */
 329 static int control_stats(struct ctdb_context *ctdb, int argc, const char **argv)
 330 {
 331         int ret;
 332         struct ctdb_statistics_wire *stats;
 333         int i, num_records = -1;
 334
 335         if (argc ==1) {
 336                 num_records = atoi(argv[0]) - 1;
 337         }
 338
 339         ret = ctdb_ctrl_getstathistory(ctdb, TIMELIMIT(), options.pnn, ctdb, &stats);
 340         if (ret != 0) {
 341                 DEBUG(DEBUG_ERR, ("Unable to get rolling statistics from node %u\n", options.pnn));
 342                 return ret;
 343         }
 344         for (i=0;i<stats->num;i++) {
 345                 if (stats->stats[i].statistics_start_time.tv_sec == 0) {
 346                         continue;
 347                 }
 348                 show_statistics(&stats->stats[i]);
 349                 if (i == num_records) {
 350                         break;
 351                 }
 352                 printf("===\n");
 353         }
 354         return 0;
 355 }
 356
 357
 358 /*
 359   display uptime of remote node
 360  */
 361 static int control_uptime(struct ctdb_context *ctdb, int argc, const char **argv)
 362 {
 363         int ret;
 364         struct ctdb_uptime *uptime = NULL;
 365         int tmp, days, hours, minutes, seconds;
 366
 367         ret = ctdb_ctrl_uptime(ctdb, ctdb, TIMELIMIT(), options.pnn, &uptime);
 368         if (ret != 0) {
 369                 DEBUG(DEBUG_ERR, ("Unable to get uptime from node %u\n", options.pnn));
 370                 return ret;
 371         }
 372
 373         if (options.machinereadable){
 374                 printf(":Current Node Time:Ctdb Start Time:Last Recovery/Failover Time:Last Recovery/IPFailover Duration:\n");
 375                 printf(":%u:%u:%u:%lf\n",
 376                         (unsigned int)uptime->current_time.tv_sec,
 377                         (unsigned int)uptime->ctdbd_start_time.tv_sec,
 378                         (unsigned int)uptime->last_recovery_finished.tv_sec,
 379                         timeval_delta(&uptime->last_recovery_finished,
 380                                       &uptime->last_recovery_started)
 381                 );
 382                 return 0;
 383         }
 384
 385         printf("Current time of node          :                %s", ctime(&uptime->current_time.tv_sec));
 386
 387         tmp = uptime->current_time.tv_sec - uptime->ctdbd_start_time.tv_sec;
 388         seconds = tmp%60;
 389         tmp    /= 60;
 390         minutes = tmp%60;
 391         tmp    /= 60;
 392         hours   = tmp%24;
 393         tmp    /= 24;
 394         days    = tmp;
 395         printf("Ctdbd start time              : (%03d %02d:%02d:%02d) %s", days, hours, minutes, seconds, ctime(&uptime->ctdbd_start_time.tv_sec));
 396
 397         tmp = uptime->current_time.tv_sec - uptime->last_recovery_finished.tv_sec;
 398         seconds = tmp%60;
 399         tmp    /= 60;
 400         minutes = tmp%60;
 401         tmp    /= 60;
 402         hours   = tmp%24;
 403         tmp    /= 24;
 404         days    = tmp;
 405         printf("Time of last recovery/failover: (%03d %02d:%02d:%02d) %s", days, hours, minutes, seconds, ctime(&uptime->last_recovery_finished.tv_sec));
 406
 407         printf("Duration of last recovery/failover: %lf seconds\n",
 408                 timeval_delta(&uptime->last_recovery_finished,
 409                               &uptime->last_recovery_started));
 410
 411         return 0;
 412 }
 413
 414 /*
 415   show the PNN of the current node
 416  */
 417 static int control_pnn(struct ctdb_context *ctdb, int argc, const char **argv)
 418 {
 419         uint32_t mypnn;
 420         bool ret;
 421
 422         ret = ctdb_getpnn(ctdb_connection, options.pnn, &mypnn);
 423         if (!ret) {
 424                 DEBUG(DEBUG_ERR, ("Unable to get pnn from node."));
 425                 return -1;
 426         }
 427
 428         printf("PNN:%d\n", mypnn);
 429         return 0;
 430 }
 431
 432
 433 struct pnn_node {
 434         struct pnn_node *next;
 435         const char *addr;
 436         int pnn;
 437 };
 438
 439 static struct pnn_node *read_nodes_file(TALLOC_CTX *mem_ctx)
 440 {
 441         const char *nodes_list;
 442         int nlines;
 443         char **lines;
 444         int i, pnn;
 445         struct pnn_node *pnn_nodes = NULL;
 446         struct pnn_node *pnn_node;
 447         struct pnn_node *tmp_node;
 448
 449         /* read the nodes file */
 450         nodes_list = getenv("CTDB_NODES");
 451         if (nodes_list == NULL) {
 452                 nodes_list = "/etc/ctdb/nodes";
 453         }
 454         lines = file_lines_load(nodes_list, &nlines, mem_ctx);
 455         if (lines == NULL) {
 456                 return NULL;
 457         }
 458         while (nlines > 0 && strcmp(lines[nlines-1], "") == 0) {
 459                 nlines--;
 460         }
 461         for (i=0, pnn=0; i<nlines; i++) {
 462                 char *node;
 463
 464                 node = lines[i];
 465                 /* strip leading spaces */
 466                 while((*node == ' ') || (*node == '\t')) {
 467                         node++;
 468                 }
 469                 if (*node == '#') {
 470                         pnn++;
 471                         continue;
 472                 }
 473                 if (strcmp(node, "") == 0) {
 474                         continue;
 475                 }
 476                 pnn_node = talloc(mem_ctx, struct pnn_node);
 477                 pnn_node->pnn = pnn++;
 478                 pnn_node->addr = talloc_strdup(pnn_node, node);
 479                 pnn_node->next = pnn_nodes;
 480                 pnn_nodes = pnn_node;
 481         }
 482
 483         /* swap them around so we return them in incrementing order */
 484         pnn_node = pnn_nodes;
 485         pnn_nodes = NULL;
 486         while (pnn_node) {
 487                 tmp_node = pnn_node;
 488                 pnn_node = pnn_node->next;
 489
 490                 tmp_node->next = pnn_nodes;
 491                 pnn_nodes = tmp_node;
 492         }
 493
 494         return pnn_nodes;
 495 }
 496
 497 /*
 498   show the PNN of the current node
 499   discover the pnn by loading the nodes file and try to bind to all
 500   addresses one at a time until the ip address is found.
 501  */
 502 static int control_xpnn(struct ctdb_context *ctdb, int argc, const char **argv)
 503 {
 504         TALLOC_CTX *mem_ctx = talloc_new(NULL);
 505         struct pnn_node *pnn_nodes;
 506         struct pnn_node *pnn_node;
 507
 508         pnn_nodes = read_nodes_file(mem_ctx);
 509         if (pnn_nodes == NULL) {
 510                 DEBUG(DEBUG_ERR,("Failed to read nodes file\n"));
 511                 talloc_free(mem_ctx);
 512                 return -1;
 513         }
 514
 515         for(pnn_node=pnn_nodes;pnn_node;pnn_node=pnn_node->next) {
 516                 ctdb_sock_addr addr;
 517
 518                 if (parse_ip(pnn_node->addr, NULL, 63999, &addr) == 0) {
 519                         DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s' in nodes file\n", pnn_node->addr));
 520                         talloc_free(mem_ctx);
 521                         return -1;
 522                 }
 523
 524                 if (ctdb_sys_have_ip(&addr)) {
 525                         printf("PNN:%d\n", pnn_node->pnn);
 526                         talloc_free(mem_ctx);
 527                         return 0;
 528                 }
 529         }
 530
 531         printf("Failed to detect which PNN this node is\n");
 532         talloc_free(mem_ctx);
 533         return -1;
 534 }
 535
 536 /*
 537   display remote ctdb status
 538  */
 539 static int control_status(struct ctdb_context *ctdb, int argc, const char **argv)
 540 {
 541         int i, ret;
 542         struct ctdb_vnn_map *vnnmap=NULL;
 543         struct ctdb_node_map *nodemap=NULL;
 544         uint32_t recmode, recmaster;
 545         int mypnn;
 546
 547         mypnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);
 548         if (mypnn == -1) {
 549                 return -1;
 550         }
 551
 552         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap);
 553         if (ret != 0) {
 554                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
 555                 return ret;
 556         }
 557
 558         if(options.machinereadable){
 559                 printf(":Node:IP:Disconnected:Banned:Disabled:Unhealthy:Stopped:Inactive:\n");
 560                 for(i=0;i<nodemap->num;i++){
 561                         if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
 562                                 continue;
 563                         }
 564                         printf(":%d:%s:%d:%d:%d:%d:%d:%d:\n", nodemap->nodes[i].pnn,
 565                                 ctdb_addr_to_str(&nodemap->nodes[i].addr),
 566                                !!(nodemap->nodes[i].flags&NODE_FLAGS_DISCONNECTED),
 567                                !!(nodemap->nodes[i].flags&NODE_FLAGS_BANNED),
 568                                !!(nodemap->nodes[i].flags&NODE_FLAGS_PERMANENTLY_DISABLED),
 569                                !!(nodemap->nodes[i].flags&NODE_FLAGS_UNHEALTHY),
 570                                !!(nodemap->nodes[i].flags&NODE_FLAGS_STOPPED),
 571                                !!(nodemap->nodes[i].flags&NODE_FLAGS_INACTIVE));
 572                 }
 573                 return 0;
 574         }
 575
 576         printf("Number of nodes:%d\n", nodemap->num);
 577         for(i=0;i<nodemap->num;i++){
 578                 static const struct {
 579                         uint32_t flag;
 580                         const char *name;
 581                 } flag_names[] = {
 582                         { NODE_FLAGS_DISCONNECTED,          "DISCONNECTED" },
 583                         { NODE_FLAGS_PERMANENTLY_DISABLED,  "DISABLED" },
 584                         { NODE_FLAGS_BANNED,                "BANNED" },
 585                         { NODE_FLAGS_UNHEALTHY,             "UNHEALTHY" },
 586                         { NODE_FLAGS_DELETED,               "DELETED" },
 587                         { NODE_FLAGS_STOPPED,               "STOPPED" },
 588                         { NODE_FLAGS_INACTIVE,              "INACTIVE" },
 589                 };
 590                 char *flags_str = NULL;
 591                 int j;
 592
 593                 if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
 594                         continue;
 595                 }
 596                 for (j=0;j<ARRAY_SIZE(flag_names);j++) {
 597                         if (nodemap->nodes[i].flags & flag_names[j].flag) {
 598                                 if (flags_str == NULL) {
 599                                         flags_str = talloc_strdup(ctdb, flag_names[j].name);
 600                                 } else {
 601                                         flags_str = talloc_asprintf_append(flags_str, "|%s",
 602                                                                            flag_names[j].name);
 603                                 }
 604                                 CTDB_NO_MEMORY_FATAL(ctdb, flags_str);
 605                         }
 606                 }
 607                 if (flags_str == NULL) {
 608                         flags_str = talloc_strdup(ctdb, "OK");
 609                         CTDB_NO_MEMORY_FATAL(ctdb, flags_str);
 610                 }
 611                 printf("pnn:%d %-16s %s%s\n", nodemap->nodes[i].pnn,
 612                        ctdb_addr_to_str(&nodemap->nodes[i].addr),
 613                        flags_str,
 614                        nodemap->nodes[i].pnn == mypnn?" (THIS NODE)":"");
 615                 talloc_free(flags_str);
 616         }
 617
 618         ret = ctdb_ctrl_getvnnmap(ctdb, TIMELIMIT(), options.pnn, ctdb, &vnnmap);
 619         if (ret != 0) {
 620                 DEBUG(DEBUG_ERR, ("Unable to get vnnmap from node %u\n", options.pnn));
 621                 return ret;
 622         }
 623         if (vnnmap->generation == INVALID_GENERATION) {
 624                 printf("Generation:INVALID\n");
 625         } else {
 626                 printf("Generation:%d\n",vnnmap->generation);
 627         }
 628         printf("Size:%d\n",vnnmap->size);
 629         for(i=0;i<vnnmap->size;i++){
 630                 printf("hash:%d lmaster:%d\n", i, vnnmap->map[i]);
 631         }
 632
 633         ret = ctdb_ctrl_getrecmode(ctdb, ctdb, TIMELIMIT(), options.pnn, &recmode);
 634         if (ret != 0) {
 635                 DEBUG(DEBUG_ERR, ("Unable to get recmode from node %u\n", options.pnn));
 636                 return ret;
 637         }
 638         printf("Recovery mode:%s (%d)\n",recmode==CTDB_RECOVERY_NORMAL?"NORMAL":"RECOVERY",recmode);
 639
 640         ret = ctdb_ctrl_getrecmaster(ctdb, ctdb, TIMELIMIT(), options.pnn, &recmaster);
 641         if (ret != 0) {
 642                 DEBUG(DEBUG_ERR, ("Unable to get recmaster from node %u\n", options.pnn));
 643                 return ret;
 644         }
 645         printf("Recovery master:%d\n",recmaster);
 646
 647         return 0;
 648 }
 649
 650
 651 struct natgw_node {
 652         struct natgw_node *next;
 653         const char *addr;
 654 };
 655
 656 /*
 657   display the list of nodes belonging to this natgw configuration
 658  */
 659 static int control_natgwlist(struct ctdb_context *ctdb, int argc, const char **argv)
 660 {
 661         int i, ret;
 662         uint32_t capabilities;
 663         const char *natgw_list;
 664         int nlines;
 665         char **lines;
 666         struct natgw_node *natgw_nodes = NULL;
 667         struct natgw_node *natgw_node;
 668         struct ctdb_node_map *nodemap=NULL;
 669
 670
 671         /* read the natgw nodes file into a linked list */
 672         natgw_list = getenv("NATGW_NODES");
 673         if (natgw_list == NULL) {
 674                 natgw_list = "/etc/ctdb/natgw_nodes";
 675         }
 676         lines = file_lines_load(natgw_list, &nlines, ctdb);
 677         if (lines == NULL) {
 678                 ctdb_set_error(ctdb, "Failed to load natgw node list '%s'\n", natgw_list);
 679                 return -1;
 680         }
 681         while (nlines > 0 && strcmp(lines[nlines-1], "") == 0) {
 682                 nlines--;
 683         }
 684         for (i=0;i<nlines;i++) {
 685                 char *node;
 686
 687                 node = lines[i];
 688                 /* strip leading spaces */
 689                 while((*node == ' ') || (*node == '\t')) {
 690                         node++;
 691                 }
 692                 if (*node == '#') {
 693                         continue;
 694                 }
 695                 if (strcmp(node, "") == 0) {
 696                         continue;
 697                 }
 698                 natgw_node = talloc(ctdb, struct natgw_node);
 699                 natgw_node->addr = talloc_strdup(natgw_node, node);
 700                 CTDB_NO_MEMORY(ctdb, natgw_node->addr);
 701                 natgw_node->next = natgw_nodes;
 702                 natgw_nodes = natgw_node;
 703         }
 704
 705         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap);
 706         if (ret != 0) {
 707                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node.\n"));
 708                 return ret;
 709         }
 710
 711         i=0;
 712         while(i<nodemap->num) {
 713                 for(natgw_node=natgw_nodes;natgw_node;natgw_node=natgw_node->next) {
 714                         if (!strcmp(natgw_node->addr, ctdb_addr_to_str(&nodemap->nodes[i].addr))) {
 715                                 break;
 716                         }
 717                 }
 718
 719                 /* this node was not in the natgw so we just remove it from
 720                  * the list
 721                  */
 722                 if ((natgw_node == NULL)
 723                 ||  (nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED) ) {
 724                         int j;
 725
 726                         for (j=i+1; j<nodemap->num; j++) {
 727                                 nodemap->nodes[j-1] = nodemap->nodes[j];
 728                         }
 729                         nodemap->num--;
 730                         continue;
 731                 }
 732
 733                 i++;
 734         }
 735
 736         /* pick a node to be natgwmaster
 737          * we dont allow STOPPED, DELETED, BANNED or UNHEALTHY nodes to become the natgwmaster
 738          */
 739         for(i=0;i<nodemap->num;i++){
 740                 if (!(nodemap->nodes[i].flags & (NODE_FLAGS_DISCONNECTED|NODE_FLAGS_STOPPED|NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_UNHEALTHY))) {
 741                         ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, &capabilities);
 742                         if (ret != 0) {
 743                                 DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", nodemap->nodes[i].pnn));
 744                                 return ret;
 745                         }
 746                         if (!(capabilities&CTDB_CAP_NATGW)) {
 747                                 continue;
 748                         }
 749                         printf("%d %s\n", nodemap->nodes[i].pnn,ctdb_addr_to_str(&nodemap->nodes[i].addr));
 750                         break;
 751                 }
 752         }
 753         /* we couldnt find any healthy node, try unhealthy ones */
 754         if (i == nodemap->num) {
 755                 for(i=0;i<nodemap->num;i++){
 756                         if (!(nodemap->nodes[i].flags & (NODE_FLAGS_DISCONNECTED|NODE_FLAGS_STOPPED|NODE_FLAGS_DELETED))) {
 757                                 ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, &capabilities);
 758                                 if (ret != 0) {
 759                                         DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", nodemap->nodes[i].pnn));
 760                                         return ret;
 761                                 }
 762                                 if (!(capabilities&CTDB_CAP_NATGW)) {
 763                                         continue;
 764                                 }
 765                                 printf("%d %s\n", nodemap->nodes[i].pnn,ctdb_addr_to_str(&nodemap->nodes[i].addr));
 766                                 break;
 767                         }
 768                 }
 769         }
 770         /* unless all nodes are STOPPED, when we pick one anyway */
 771         if (i == nodemap->num) {
 772                 for(i=0;i<nodemap->num;i++){
 773                         if (!(nodemap->nodes[i].flags & (NODE_FLAGS_DISCONNECTED|NODE_FLAGS_DELETED))) {
 774                                 ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, &capabilities);
 775                                 if (ret != 0) {
 776                                         DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", nodemap->nodes[i].pnn));
 777                                         return ret;
 778                                 }
 779                                 if (!(capabilities&CTDB_CAP_NATGW)) {
 780                                         continue;
 781                                 }
 782                                 printf("%d %s\n", nodemap->nodes[i].pnn, ctdb_addr_to_str(&nodemap->nodes[i].addr));
 783                                 break;
 784                         }
 785                 }
 786                 /* or if we still can not find any */
 787                 if (i == nodemap->num) {
 788                         printf("-1 0.0.0.0\n");
 789                 }
 790         }
 791
 792         /* print the pruned list of nodes belonging to this natgw list */
 793         for(i=0;i<nodemap->num;i++){
 794                 if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
 795                         continue;
 796                 }
 797                 printf(":%d:%s:%d:%d:%d:%d:%d\n", nodemap->nodes[i].pnn,
 798                         ctdb_addr_to_str(&nodemap->nodes[i].addr),
 799                        !!(nodemap->nodes[i].flags&NODE_FLAGS_DISCONNECTED),
 800                        !!(nodemap->nodes[i].flags&NODE_FLAGS_BANNED),
 801                        !!(nodemap->nodes[i].flags&NODE_FLAGS_PERMANENTLY_DISABLED),
 802                        !!(nodemap->nodes[i].flags&NODE_FLAGS_UNHEALTHY),
 803                        !!(nodemap->nodes[i].flags&NODE_FLAGS_STOPPED));
 804         }
 805
 806         return 0;
 807 }
 808
 809 /*
 810   display the status of the scripts for monitoring (or other events)
 811  */
 812 static int control_one_scriptstatus(struct ctdb_context *ctdb,
 813                                     enum ctdb_eventscript_call type)
 814 {
 815         struct ctdb_scripts_wire *script_status;
 816         int ret, i;
 817
 818         ret = ctdb_ctrl_getscriptstatus(ctdb, TIMELIMIT(), options.pnn, ctdb, type, &script_status);
 819         if (ret != 0) {
 820                 DEBUG(DEBUG_ERR, ("Unable to get script status from node %u\n", options.pnn));
 821                 return ret;
 822         }
 823
 824         if (script_status == NULL) {
 825                 if (!options.machinereadable) {
 826                         printf("%s cycle never run\n",
 827                                ctdb_eventscript_call_names[type]);
 828                 }
 829                 return 0;
 830         }
 831
 832         if (!options.machinereadable) {
 833                 printf("%d scripts were executed last %s cycle\n",
 834                        script_status->num_scripts,
 835                        ctdb_eventscript_call_names[type]);
 836         }
 837         for (i=0; i<script_status->num_scripts; i++) {
 838                 const char *status = NULL;
 839
 840                 switch (script_status->scripts[i].status) {
 841                 case -ETIME:
 842                         status = "TIMEDOUT";
 843                         break;
 844                 case -ENOEXEC:
 845                         status = "DISABLED";
 846                         break;
 847                 case 0:
 848                         status = "OK";
 849                         break;
 850                 default:
 851                         if (script_status->scripts[i].status > 0)
 852                                 status = "ERROR";
 853                         break;
 854                 }
 855                 if (options.machinereadable) {
 856                         printf("%s:%s:%i:%s:%lu.%06lu:%lu.%06lu:%s:\n",
 857                                ctdb_eventscript_call_names[type],
 858                                script_status->scripts[i].name,
 859                                script_status->scripts[i].status,
 860                                status,
 861                                (long)script_status->scripts[i].start.tv_sec,
 862                                (long)script_status->scripts[i].start.tv_usec,
 863                                (long)script_status->scripts[i].finished.tv_sec,
 864                                (long)script_status->scripts[i].finished.tv_usec,
 865                                script_status->scripts[i].output);
 866                         continue;
 867                 }
 868                 if (status)
 869                         printf("%-20s Status:%s    ",
 870                                script_status->scripts[i].name, status);
 871                 else
 872                         /* Some other error, eg from stat. */
 873                         printf("%-20s Status:CANNOT RUN (%s)",
 874                                script_status->scripts[i].name,
 875                                strerror(-script_status->scripts[i].status));
 876
 877                 if (script_status->scripts[i].status >= 0) {
 878                         printf("Duration:%.3lf ",
 879                         timeval_delta(&script_status->scripts[i].finished,
 880                               &script_status->scripts[i].start));
 881                 }
 882                 if (script_status->scripts[i].status != -ENOEXEC) {
 883                         printf("%s",
 884                                ctime(&script_status->scripts[i].start.tv_sec));
 885                         if (script_status->scripts[i].status != 0) {
 886                                 printf("   OUTPUT:%s\n",
 887                                        script_status->scripts[i].output);
 888                         }
 889                 } else {
 890                         printf("\n");
 891                 }
 892         }
 893         return 0;
 894 }
 895
 896
 897 static int control_scriptstatus(struct ctdb_context *ctdb,
 898                                 int argc, const char **argv)
 899 {
 900         int ret;
 901         enum ctdb_eventscript_call type, min, max;
 902         const char *arg;
 903
 904         if (argc > 1) {
 905                 DEBUG(DEBUG_ERR, ("Unknown arguments to scriptstatus\n"));
 906                 return -1;
 907         }
 908
 909         if (argc == 0)
 910                 arg = ctdb_eventscript_call_names[CTDB_EVENT_MONITOR];
 911         else
 912                 arg = argv[0];
 913
 914         for (type = 0; type < CTDB_EVENT_MAX; type++) {
 915                 if (strcmp(arg, ctdb_eventscript_call_names[type]) == 0) {
 916                         min = type;
 917                         max = type+1;
 918                         break;
 919                 }
 920         }
 921         if (type == CTDB_EVENT_MAX) {
 922                 if (strcmp(arg, "all") == 0) {
 923                         min = 0;
 924                         max = CTDB_EVENT_MAX;
 925                 } else {
 926                         DEBUG(DEBUG_ERR, ("Unknown event type %s\n", argv[0]));
 927                         return -1;
 928                 }
 929         }
 930
 931         if (options.machinereadable) {
 932                 printf(":Type:Name:Code:Status:Start:End:Error Output...:\n");
 933         }
 934
 935         for (type = min; type < max; type++) {
 936                 ret = control_one_scriptstatus(ctdb, type);
 937                 if (ret != 0) {
 938                         return ret;
 939                 }
 940         }
 941
 942         return 0;
 943 }
 944
 945 /*
 946   enable an eventscript
 947  */
 948 static int control_enablescript(struct ctdb_context *ctdb, int argc, const char **argv)
 949 {
 950         int ret;
 951
 952         if (argc < 1) {
 953                 usage();
 954         }
 955
 956         ret = ctdb_ctrl_enablescript(ctdb, TIMELIMIT(), options.pnn, argv[0]);
 957         if (ret != 0) {
 958           DEBUG(DEBUG_ERR, ("Unable to enable script %s on node %u\n", argv[0], options.pnn));
 959                 return ret;
 960         }
 961
 962         return 0;
 963 }
 964
 965 /*
 966   disable an eventscript
 967  */
 968 static int control_disablescript(struct ctdb_context *ctdb, int argc, const char **argv)
 969 {
 970         int ret;
 971
 972         if (argc < 1) {
 973                 usage();
 974         }
 975
 976         ret = ctdb_ctrl_disablescript(ctdb, TIMELIMIT(), options.pnn, argv[0]);
 977         if (ret != 0) {
 978           DEBUG(DEBUG_ERR, ("Unable to disable script %s on node %u\n", argv[0], options.pnn));
 979                 return ret;
 980         }
 981
 982         return 0;
 983 }
 984
 985 /*
 986   display the pnn of the recovery master
 987  */
 988 static int control_recmaster(struct ctdb_context *ctdb, int argc, const char **argv)
 989 {
 990         int ret;
 991         uint32_t recmaster;
 992
 993         ret = ctdb_ctrl_getrecmaster(ctdb, ctdb, TIMELIMIT(), options.pnn, &recmaster);
 994         if (ret != 0) {
 995                 DEBUG(DEBUG_ERR, ("Unable to get recmaster from node %u\n", options.pnn));
 996                 return ret;
 997         }
 998         printf("%d\n",recmaster);
 999
1000         return 0;
1001 }
1002
1003 /*
1004   add a tickle to a public address
1005  */
1006 static int control_add_tickle(struct ctdb_context *ctdb, int argc, const char **argv)
1007 {
1008         struct ctdb_tcp_connection t;
1009         TDB_DATA data;
1010         int ret;
1011
1012         if (argc < 2) {
1013                 usage();
1014         }
1015
1016         if (parse_ip_port(argv[0], &t.src_addr) == 0) {
1017                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
1018                 return -1;
1019         }
1020         if (parse_ip_port(argv[1], &t.dst_addr) == 0) {
1021                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[1]));
1022                 return -1;
1023         }
1024
1025         data.dptr = (uint8_t *)&t;
1026         data.dsize = sizeof(t);
1027
1028         /* tell all nodes about this tcp connection */
1029         ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_TCP_ADD_DELAYED_UPDATE,
1030                            0, data, ctdb, NULL, NULL, NULL, NULL);
1031         if (ret != 0) {
1032                 DEBUG(DEBUG_ERR,("Failed to add tickle\n"));
1033                 return -1;
1034         }
1035
1036         return 0;
1037 }
1038
1039
1040 /*
1041   delete a tickle from a node
1042  */
1043 static int control_del_tickle(struct ctdb_context *ctdb, int argc, const char **argv)
1044 {
1045         struct ctdb_tcp_connection t;
1046         TDB_DATA data;
1047         int ret;
1048
1049         if (argc < 2) {
1050                 usage();
1051         }
1052
1053         if (parse_ip_port(argv[0], &t.src_addr) == 0) {
1054                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
1055                 return -1;
1056         }
1057         if (parse_ip_port(argv[1], &t.dst_addr) == 0) {
1058                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[1]));
1059                 return -1;
1060         }
1061
1062         data.dptr = (uint8_t *)&t;
1063         data.dsize = sizeof(t);
1064
1065         /* tell all nodes about this tcp connection */
1066         ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_TCP_REMOVE,
1067                            0, data, ctdb, NULL, NULL, NULL, NULL);
1068         if (ret != 0) {
1069                 DEBUG(DEBUG_ERR,("Failed to remove tickle\n"));
1070                 return -1;
1071         }
1072
1073         return 0;
1074 }
1075
1076
1077 /*
1078   get a list of all tickles for this pnn
1079  */
1080 static int control_get_tickles(struct ctdb_context *ctdb, int argc, const char **argv)
1081 {
1082         struct ctdb_control_tcp_tickle_list *list;
1083         ctdb_sock_addr addr;
1084         int i, ret;
1085         unsigned port = 0;
1086
1087         if (argc < 1) {
1088                 usage();
1089         }
1090
1091         if (argc == 2) {
1092                 port = atoi(argv[1]);
1093         }
1094
1095         if (parse_ip(argv[0], NULL, 0, &addr) == 0) {
1096                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
1097                 return -1;
1098         }
1099
1100         ret = ctdb_ctrl_get_tcp_tickles(ctdb, TIMELIMIT(), options.pnn, ctdb, &addr, &list);
1101         if (ret == -1) {
1102                 DEBUG(DEBUG_ERR, ("Unable to list tickles\n"));
1103                 return -1;
1104         }
1105
1106         if (options.machinereadable){
1107                 printf(":source ip:port:destination ip:port:\n");
1108                 for (i=0;i<list->tickles.num;i++) {
1109                         if (port && port != ntohs(list->tickles.connections[i].dst_addr.ip.sin_port)) {
1110                                 continue;
1111                         }
1112                         printf(":%s:%u", ctdb_addr_to_str(&list->tickles.connections[i].src_addr), ntohs(list->tickles.connections[i].src_addr.ip.sin_port));
1113                         printf(":%s:%u:\n", ctdb_addr_to_str(&list->tickles.connections[i].dst_addr), ntohs(list->tickles.connections[i].dst_addr.ip.sin_port));
1114                 }
1115         } else {
1116                 printf("Tickles for ip:%s\n", ctdb_addr_to_str(&list->addr));
1117                 printf("Num tickles:%u\n", list->tickles.num);
1118                 for (i=0;i<list->tickles.num;i++) {
1119                         if (port && port != ntohs(list->tickles.connections[i].dst_addr.ip.sin_port)) {
1120                                 continue;
1121                         }
1122                         printf("SRC: %s:%u   ", ctdb_addr_to_str(&list->tickles.connections[i].src_addr), ntohs(list->tickles.connections[i].src_addr.ip.sin_port));
1123                         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));
1124                 }
1125         }
1126
1127         talloc_free(list);
1128
1129         return 0;
1130 }
1131
1132
1133 static int move_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr, uint32_t pnn)
1134 {
1135         struct ctdb_all_public_ips *ips;
1136         struct ctdb_public_ip ip;
1137         int i, ret;
1138         uint32_t *nodes;
1139         uint32_t disable_time;
1140         TDB_DATA data;
1141         struct ctdb_node_map *nodemap=NULL;
1142         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1143
1144         disable_time = 30;
1145         data.dptr  = (uint8_t*)&disable_time;
1146         data.dsize = sizeof(disable_time);
1147         ret = ctdb_client_send_message(ctdb, CTDB_BROADCAST_CONNECTED, CTDB_SRVID_DISABLE_IP_CHECK, data);
1148         if (ret != 0) {
1149                 DEBUG(DEBUG_ERR,("Failed to send message to disable ipcheck\n"));
1150                 return -1;
1151         }
1152
1153
1154
1155         /* read the public ip list from the node */
1156         ret = ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), pnn, ctdb, &ips);
1157         if (ret != 0) {
1158                 DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %u\n", pnn));
1159                 talloc_free(tmp_ctx);
1160                 return -1;
1161         }
1162
1163         for (i=0;i<ips->num;i++) {
1164                 if (ctdb_same_ip(addr, &ips->ips[i].addr)) {
1165                         break;
1166                 }
1167         }
1168         if (i==ips->num) {
1169                 DEBUG(DEBUG_ERR, ("Node %u can not host ip address '%s'\n",
1170                         pnn, ctdb_addr_to_str(addr)));
1171                 talloc_free(tmp_ctx);
1172                 return -1;
1173         }
1174
1175         ip.pnn  = pnn;
1176         ip.addr = *addr;
1177
1178         data.dptr  = (uint8_t *)&ip;
1179         data.dsize = sizeof(ip);
1180
1181         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, tmp_ctx, &nodemap);
1182         if (ret != 0) {
1183                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
1184                 talloc_free(tmp_ctx);
1185                 return ret;
1186         }
1187
1188         nodes = list_of_active_nodes_except_pnn(ctdb, nodemap, tmp_ctx, pnn);
1189         ret = ctdb_client_async_control(ctdb, CTDB_CONTROL_RELEASE_IP,
1190                                         nodes, 0,
1191                                         LONGTIMELIMIT(),
1192                                         false, data,
1193                                         NULL, NULL,
1194                                         NULL);
1195         if (ret != 0) {
1196                 DEBUG(DEBUG_ERR,("Failed to release IP on nodes\n"));
1197                 talloc_free(tmp_ctx);
1198                 return -1;
1199         }
1200
1201         ret = ctdb_ctrl_takeover_ip(ctdb, LONGTIMELIMIT(), pnn, &ip);
1202         if (ret != 0) {
1203                 DEBUG(DEBUG_ERR,("Failed to take over IP on node %d\n", pnn));
1204                 talloc_free(tmp_ctx);
1205                 return -1;
1206         }
1207
1208         /* update the recovery daemon so it now knows to expect the new
1209            node assignment for this ip.
1210         */
1211         ret = ctdb_client_send_message(ctdb, CTDB_BROADCAST_CONNECTED, CTDB_SRVID_RECD_UPDATE_IP, data);
1212         if (ret != 0) {
1213                 DEBUG(DEBUG_ERR,("Failed to send message to update the ip on the recovery master.\n"));
1214                 return -1;
1215         }
1216
1217         talloc_free(tmp_ctx);
1218         return 0;
1219 }
1220
1221 /*
1222   move/failover an ip address to a specific node
1223  */
1224 static int control_moveip(struct ctdb_context *ctdb, int argc, const char **argv)
1225 {
1226         uint32_t pnn;
1227         ctdb_sock_addr addr;
1228
1229         if (argc < 2) {
1230                 usage();
1231                 return -1;
1232         }
1233
1234         if (parse_ip(argv[0], NULL, 0, &addr) == 0) {
1235                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
1236                 return -1;
1237         }
1238
1239
1240         if (sscanf(argv[1], "%u", &pnn) != 1) {
1241                 DEBUG(DEBUG_ERR, ("Badly formed pnn\n"));
1242                 return -1;
1243         }
1244
1245         if (move_ip(ctdb, &addr, pnn) != 0) {
1246                 DEBUG(DEBUG_ERR,("Failed to move ip to node %d\n", pnn));
1247                 return -1;
1248         }
1249
1250         return 0;
1251 }
1252
1253 void getips_store_callback(void *param, void *data)
1254 {
1255         struct ctdb_public_ip *node_ip = (struct ctdb_public_ip *)data;
1256         struct ctdb_all_public_ips *ips = param;
1257         int i;
1258
1259         i = ips->num++;
1260         ips->ips[i].pnn  = node_ip->pnn;
1261         ips->ips[i].addr = node_ip->addr;
1262 }
1263
1264 void getips_count_callback(void *param, void *data)
1265 {
1266         uint32_t *count = param;
1267
1268         (*count)++;
1269 }
1270
1271 #define IP_KEYLEN       4
1272 static uint32_t *ip_key(ctdb_sock_addr *ip)
1273 {
1274         static uint32_t key[IP_KEYLEN];
1275
1276         bzero(key, sizeof(key));
1277
1278         switch (ip->sa.sa_family) {
1279         case AF_INET:
1280                 key[0]  = ip->ip.sin_addr.s_addr;
1281                 break;
1282         case AF_INET6:
1283                 key[0]  = ip->ip6.sin6_addr.s6_addr32[3];
1284                 key[1]  = ip->ip6.sin6_addr.s6_addr32[2];
1285                 key[2]  = ip->ip6.sin6_addr.s6_addr32[1];
1286                 key[3]  = ip->ip6.sin6_addr.s6_addr32[0];
1287                 break;
1288         default:
1289                 DEBUG(DEBUG_ERR, (__location__ " ERROR, unknown family passed :%u\n", ip->sa.sa_family));
1290                 return key;
1291         }
1292
1293         return key;
1294 }
1295
1296 static void *add_ip_callback(void *parm, void *data)
1297 {
1298         return parm;
1299 }
1300
1301 static int
1302 control_get_all_public_ips(struct ctdb_context *ctdb, TALLOC_CTX *tmp_ctx, struct ctdb_all_public_ips **ips)
1303 {
1304         struct ctdb_all_public_ips *tmp_ips;
1305         struct ctdb_node_map *nodemap=NULL;
1306         trbt_tree_t *ip_tree;
1307         int i, j, len, ret;
1308         uint32_t count;
1309
1310         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, tmp_ctx, &nodemap);
1311         if (ret != 0) {
1312                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
1313                 return ret;
1314         }
1315
1316         ip_tree = trbt_create(tmp_ctx, 0);
1317
1318         for(i=0;i<nodemap->num;i++){
1319                 if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
1320                         continue;
1321                 }
1322                 if (nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED) {
1323                         continue;
1324                 }
1325
1326                 /* read the public ip list from this node */
1327                 ret = ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, tmp_ctx, &tmp_ips);
1328                 if (ret != 0) {
1329                         DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %u\n", nodemap->nodes[i].pnn));
1330                         return -1;
1331                 }
1332
1333                 for (j=0; j<tmp_ips->num;j++) {
1334                         struct ctdb_public_ip *node_ip;
1335
1336                         node_ip = talloc(tmp_ctx, struct ctdb_public_ip);
1337                         node_ip->pnn  = tmp_ips->ips[j].pnn;
1338                         node_ip->addr = tmp_ips->ips[j].addr;
1339
1340                         trbt_insertarray32_callback(ip_tree,
1341                                 IP_KEYLEN, ip_key(&tmp_ips->ips[j].addr),
1342                                 add_ip_callback,
1343                                 node_ip);
1344                 }
1345                 talloc_free(tmp_ips);
1346         }
1347
1348         /* traverse */
1349         count = 0;
1350         trbt_traversearray32(ip_tree, IP_KEYLEN, getips_count_callback, &count);
1351
1352         len = offsetof(struct ctdb_all_public_ips, ips) +
1353                 count*sizeof(struct ctdb_public_ip);
1354         tmp_ips = talloc_zero_size(tmp_ctx, len);
1355         trbt_traversearray32(ip_tree, IP_KEYLEN, getips_store_callback, tmp_ips);
1356
1357         *ips = tmp_ips;
1358
1359         return 0;
1360 }
1361
1362
1363 /*
1364  * scans all other nodes and returns a pnn for another node that can host this
1365  * ip address or -1
1366  */
1367 static int
1368 find_other_host_for_public_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr)
1369 {
1370         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1371         struct ctdb_all_public_ips *ips;
1372         struct ctdb_node_map *nodemap=NULL;
1373         int i, j, ret;
1374
1375         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, tmp_ctx, &nodemap);
1376         if (ret != 0) {
1377                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
1378                 talloc_free(tmp_ctx);
1379                 return ret;
1380         }
1381
1382         for(i=0;i<nodemap->num;i++){
1383                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
1384                         continue;
1385                 }
1386                 if (nodemap->nodes[i].pnn == options.pnn) {
1387                         continue;
1388                 }
1389
1390                 /* read the public ip list from this node */
1391                 ret = ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, tmp_ctx, &ips);
1392                 if (ret != 0) {
1393                         DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %u\n", nodemap->nodes[i].pnn));
1394                         return -1;
1395                 }
1396
1397                 for (j=0;j<ips->num;j++) {
1398                         if (ctdb_same_ip(addr, &ips->ips[j].addr)) {
1399                                 talloc_free(tmp_ctx);
1400                                 return nodemap->nodes[i].pnn;
1401                         }
1402                 }
1403                 talloc_free(ips);
1404         }
1405
1406         talloc_free(tmp_ctx);
1407         return -1;
1408 }
1409
1410 /*
1411   add a public ip address to a node
1412  */
1413 static int control_addip(struct ctdb_context *ctdb, int argc, const char **argv)
1414 {
1415         int i, ret;
1416         int len;
1417         uint32_t pnn;
1418         unsigned mask;
1419         ctdb_sock_addr addr;
1420         struct ctdb_control_ip_iface *pub;
1421         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1422         struct ctdb_all_public_ips *ips;
1423
1424
1425         if (argc != 2) {
1426                 talloc_free(tmp_ctx);
1427                 usage();
1428         }
1429
1430         if (!parse_ip_mask(argv[0], argv[1], &addr, &mask)) {
1431                 DEBUG(DEBUG_ERR, ("Badly formed ip/mask : %s\n", argv[0]));
1432                 talloc_free(tmp_ctx);
1433                 return -1;
1434         }
1435
1436         ret = control_get_all_public_ips(ctdb, tmp_ctx, &ips);
1437         if (ret != 0) {
1438                 DEBUG(DEBUG_ERR, ("Unable to get public ip list from cluster\n"));
1439                 talloc_free(tmp_ctx);
1440                 return ret;
1441         }
1442
1443
1444         /* check if some other node is already serving this ip, if not,
1445          * we will claim it
1446          */
1447         for (i=0;i<ips->num;i++) {
1448                 if (ctdb_same_ip(&addr, &ips->ips[i].addr)) {
1449                         break;
1450                 }
1451         }
1452
1453         len = offsetof(struct ctdb_control_ip_iface, iface) + strlen(argv[1]) + 1;
1454         pub = talloc_size(tmp_ctx, len);
1455         CTDB_NO_MEMORY(ctdb, pub);
1456
1457         pub->addr  = addr;
1458         pub->mask  = mask;
1459         pub->len   = strlen(argv[1])+1;
1460         memcpy(&pub->iface[0], argv[1], strlen(argv[1])+1);
1461
1462         ret = ctdb_ctrl_add_public_ip(ctdb, TIMELIMIT(), options.pnn, pub);
1463         if (ret != 0) {
1464                 DEBUG(DEBUG_ERR, ("Unable to add public ip to node %u\n", options.pnn));
1465                 talloc_free(tmp_ctx);
1466                 return ret;
1467         }
1468
1469         if (i == ips->num) {
1470                 /* no one has this ip so we claim it */
1471                 pnn  = options.pnn;
1472         } else {
1473                 pnn  = ips->ips[i].pnn;
1474         }
1475
1476         if (move_ip(ctdb, &addr, pnn) != 0) {
1477                 DEBUG(DEBUG_ERR,("Failed to move ip to node %d\n", pnn));
1478                 return -1;
1479         }
1480
1481         talloc_free(tmp_ctx);
1482         return 0;
1483 }
1484
1485 static int control_delip(struct ctdb_context *ctdb, int argc, const char **argv);
1486
1487 static int control_delip_all(struct ctdb_context *ctdb, int argc, const char **argv, ctdb_sock_addr *addr)
1488 {
1489         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1490         struct ctdb_node_map *nodemap=NULL;
1491         struct ctdb_all_public_ips *ips;
1492         int ret, i, j;
1493
1494         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, tmp_ctx, &nodemap);
1495         if (ret != 0) {
1496                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from current node\n"));
1497                 return ret;
1498         }
1499
1500         /* remove it from the nodes that are not hosting the ip currently */
1501         for(i=0;i<nodemap->num;i++){
1502                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
1503                         continue;
1504                 }
1505                 if (ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, tmp_ctx, &ips) != 0) {
1506                         DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %d\n", nodemap->nodes[i].pnn));
1507                         continue;
1508                 }
1509
1510                 for (j=0;j<ips->num;j++) {
1511                         if (ctdb_same_ip(addr, &ips->ips[j].addr)) {
1512                                 break;
1513                         }
1514                 }
1515                 if (j==ips->num) {
1516                         continue;
1517                 }
1518
1519                 if (ips->ips[j].pnn == nodemap->nodes[i].pnn) {
1520                         continue;
1521                 }
1522
1523                 options.pnn = nodemap->nodes[i].pnn;
1524                 control_delip(ctdb, argc, argv);
1525         }
1526
1527
1528         /* remove it from every node (also the one hosting it) */
1529         for(i=0;i<nodemap->num;i++){
1530                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
1531                         continue;
1532                 }
1533                 if (ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, tmp_ctx, &ips) != 0) {
1534                         DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %d\n", nodemap->nodes[i].pnn));
1535                         continue;
1536                 }
1537
1538                 for (j=0;j<ips->num;j++) {
1539                         if (ctdb_same_ip(addr, &ips->ips[j].addr)) {
1540                                 break;
1541                         }
1542                 }
1543                 if (j==ips->num) {
1544                         continue;
1545                 }
1546
1547                 options.pnn = nodemap->nodes[i].pnn;
1548                 control_delip(ctdb, argc, argv);
1549         }
1550
1551         talloc_free(tmp_ctx);
1552         return 0;
1553 }
1554
1555 /*
1556   delete a public ip address from a node
1557  */
1558 static int control_delip(struct ctdb_context *ctdb, int argc, const char **argv)
1559 {
1560         int i, ret;
1561         ctdb_sock_addr addr;
1562         struct ctdb_control_ip_iface pub;
1563         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1564         struct ctdb_all_public_ips *ips;
1565
1566         if (argc != 1) {
1567                 talloc_free(tmp_ctx);
1568                 usage();
1569         }
1570
1571         if (parse_ip(argv[0], NULL, 0, &addr) == 0) {
1572                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
1573                 return -1;
1574         }
1575
1576         if (options.pnn == CTDB_BROADCAST_ALL) {
1577                 return control_delip_all(ctdb, argc, argv, &addr);
1578         }
1579
1580         pub.addr  = addr;
1581         pub.mask  = 0;
1582         pub.len   = 0;
1583
1584         ret = ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), options.pnn, tmp_ctx, &ips);
1585         if (ret != 0) {
1586                 DEBUG(DEBUG_ERR, ("Unable to get public ip list from cluster\n"));
1587                 talloc_free(tmp_ctx);
1588                 return ret;
1589         }
1590
1591         for (i=0;i<ips->num;i++) {
1592                 if (ctdb_same_ip(&addr, &ips->ips[i].addr)) {
1593                         break;
1594                 }
1595         }
1596
1597         if (i==ips->num) {
1598                 DEBUG(DEBUG_ERR, ("This node does not support this public address '%s'\n",
1599                         ctdb_addr_to_str(&addr)));
1600                 talloc_free(tmp_ctx);
1601                 return -1;
1602         }
1603
1604         if (ips->ips[i].pnn == options.pnn) {
1605                 ret = find_other_host_for_public_ip(ctdb, &addr);
1606                 if (ret != -1) {
1607                         if (move_ip(ctdb, &addr, ret) != 0) {
1608                                 DEBUG(DEBUG_ERR,("Failed to move ip to node %d\n", ret));
1609                                 return -1;
1610                         }
1611                 }
1612         }
1613
1614         ret = ctdb_ctrl_del_public_ip(ctdb, TIMELIMIT(), options.pnn, &pub);
1615         if (ret != 0) {
1616                 DEBUG(DEBUG_ERR, ("Unable to del public ip from node %u\n", options.pnn));
1617                 talloc_free(tmp_ctx);
1618                 return ret;
1619         }
1620
1621         talloc_free(tmp_ctx);
1622         return 0;
1623 }
1624
1625 /*
1626   kill a tcp connection
1627  */
1628 static int kill_tcp(struct ctdb_context *ctdb, int argc, const char **argv)
1629 {
1630         int ret;
1631         struct ctdb_control_killtcp killtcp;
1632
1633         if (argc < 2) {
1634                 usage();
1635         }
1636
1637         if (!parse_ip_port(argv[0], &killtcp.src_addr)) {
1638                 DEBUG(DEBUG_ERR, ("Bad IP:port '%s'\n", argv[0]));
1639                 return -1;
1640         }
1641
1642         if (!parse_ip_port(argv[1], &killtcp.dst_addr)) {
1643                 DEBUG(DEBUG_ERR, ("Bad IP:port '%s'\n", argv[1]));
1644                 return -1;
1645         }
1646
1647         ret = ctdb_ctrl_killtcp(ctdb, TIMELIMIT(), options.pnn, &killtcp);
1648         if (ret != 0) {
1649                 DEBUG(DEBUG_ERR, ("Unable to killtcp from node %u\n", options.pnn));
1650                 return ret;
1651         }
1652
1653         return 0;
1654 }
1655
1656
1657 /*
1658   send a gratious arp
1659  */
1660 static int control_gratious_arp(struct ctdb_context *ctdb, int argc, const char **argv)
1661 {
1662         int ret;
1663         ctdb_sock_addr addr;
1664
1665         if (argc < 2) {
1666                 usage();
1667         }
1668
1669         if (!parse_ip(argv[0], NULL, 0, &addr)) {
1670                 DEBUG(DEBUG_ERR, ("Bad IP '%s'\n", argv[0]));
1671                 return -1;
1672         }
1673
1674         ret = ctdb_ctrl_gratious_arp(ctdb, TIMELIMIT(), options.pnn, &addr, argv[1]);
1675         if (ret != 0) {
1676                 DEBUG(DEBUG_ERR, ("Unable to send gratious_arp from node %u\n", options.pnn));
1677                 return ret;
1678         }
1679
1680         return 0;
1681 }
1682
1683 /*
1684   register a server id
1685  */
1686 static int regsrvid(struct ctdb_context *ctdb, int argc, const char **argv)
1687 {
1688         int ret;
1689         struct ctdb_server_id server_id;
1690
1691         if (argc < 3) {
1692                 usage();
1693         }
1694
1695         server_id.pnn       = strtoul(argv[0], NULL, 0);
1696         server_id.type      = strtoul(argv[1], NULL, 0);
1697         server_id.server_id = strtoul(argv[2], NULL, 0);
1698
1699         ret = ctdb_ctrl_register_server_id(ctdb, TIMELIMIT(), &server_id);
1700         if (ret != 0) {
1701                 DEBUG(DEBUG_ERR, ("Unable to register server_id from node %u\n", options.pnn));
1702                 return ret;
1703         }
1704         DEBUG(DEBUG_ERR,("Srvid registered. Sleeping for 999 seconds\n"));
1705         sleep(999);
1706         return -1;
1707 }
1708
1709 /*
1710   unregister a server id
1711  */
1712 static int unregsrvid(struct ctdb_context *ctdb, int argc, const char **argv)
1713 {
1714         int ret;
1715         struct ctdb_server_id server_id;
1716
1717         if (argc < 3) {
1718                 usage();
1719         }
1720
1721         server_id.pnn       = strtoul(argv[0], NULL, 0);
1722         server_id.type      = strtoul(argv[1], NULL, 0);
1723         server_id.server_id = strtoul(argv[2], NULL, 0);
1724
1725         ret = ctdb_ctrl_unregister_server_id(ctdb, TIMELIMIT(), &server_id);
1726         if (ret != 0) {
1727                 DEBUG(DEBUG_ERR, ("Unable to unregister server_id from node %u\n", options.pnn));
1728                 return ret;
1729         }
1730         return -1;
1731 }
1732
1733 /*
1734   check if a server id exists
1735  */
1736 static int chksrvid(struct ctdb_context *ctdb, int argc, const char **argv)
1737 {
1738         uint32_t status;
1739         int ret;
1740         struct ctdb_server_id server_id;
1741
1742         if (argc < 3) {
1743                 usage();
1744         }
1745
1746         server_id.pnn       = strtoul(argv[0], NULL, 0);
1747         server_id.type      = strtoul(argv[1], NULL, 0);
1748         server_id.server_id = strtoul(argv[2], NULL, 0);
1749
1750         ret = ctdb_ctrl_check_server_id(ctdb, TIMELIMIT(), options.pnn, &server_id, &status);
1751         if (ret != 0) {
1752                 DEBUG(DEBUG_ERR, ("Unable to check server_id from node %u\n", options.pnn));
1753                 return ret;
1754         }
1755
1756         if (status) {
1757                 printf("Server id %d:%d:%d EXISTS\n", server_id.pnn, server_id.type, server_id.server_id);
1758         } else {
1759                 printf("Server id %d:%d:%d does NOT exist\n", server_id.pnn, server_id.type, server_id.server_id);
1760         }
1761         return 0;
1762 }
1763
1764 /*
1765   get a list of all server ids that are registered on a node
1766  */
1767 static int getsrvids(struct ctdb_context *ctdb, int argc, const char **argv)
1768 {
1769         int i, ret;
1770         struct ctdb_server_id_list *server_ids;
1771
1772         ret = ctdb_ctrl_get_server_id_list(ctdb, ctdb, TIMELIMIT(), options.pnn, &server_ids);
1773         if (ret != 0) {
1774                 DEBUG(DEBUG_ERR, ("Unable to get server_id list from node %u\n", options.pnn));
1775                 return ret;
1776         }
1777
1778         for (i=0; i<server_ids->num; i++) {
1779                 printf("Server id %d:%d:%d\n",
1780                         server_ids->server_ids[i].pnn,
1781                         server_ids->server_ids[i].type,
1782                         server_ids->server_ids[i].server_id);
1783         }
1784
1785         return -1;
1786 }
1787
1788 /*
1789   send a tcp tickle ack
1790  */
1791 static int tickle_tcp(struct ctdb_context *ctdb, int argc, const char **argv)
1792 {
1793         int ret;
1794         ctdb_sock_addr  src, dst;
1795
1796         if (argc < 2) {
1797                 usage();
1798         }
1799
1800         if (!parse_ip_port(argv[0], &src)) {
1801                 DEBUG(DEBUG_ERR, ("Bad IP:port '%s'\n", argv[0]));
1802                 return -1;
1803         }
1804
1805         if (!parse_ip_port(argv[1], &dst)) {
1806                 DEBUG(DEBUG_ERR, ("Bad IP:port '%s'\n", argv[1]));
1807                 return -1;
1808         }
1809
1810         ret = ctdb_sys_send_tcp(&src, &dst, 0, 0, 0);
1811         if (ret==0) {
1812                 return 0;
1813         }
1814         DEBUG(DEBUG_ERR, ("Error while sending tickle ack\n"));
1815
1816         return -1;
1817 }
1818
1819
1820 /*
1821   display public ip status
1822  */
1823 static int control_ip(struct ctdb_context *ctdb, int argc, const char **argv)
1824 {
1825         int i, ret;
1826         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1827         struct ctdb_all_public_ips *ips;
1828
1829         if (options.pnn == CTDB_BROADCAST_ALL) {
1830                 /* read the list of public ips from all nodes */
1831                 ret = control_get_all_public_ips(ctdb, tmp_ctx, &ips);
1832         } else {
1833                 /* read the public ip list from this node */
1834                 ret = ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), options.pnn, tmp_ctx, &ips);
1835         }
1836         if (ret != 0) {
1837                 DEBUG(DEBUG_ERR, ("Unable to get public ips from node %u\n", options.pnn));
1838                 talloc_free(tmp_ctx);
1839                 return ret;
1840         }
1841
1842         if (options.machinereadable){
1843                 printf(":Public IP:Node:");
1844                 if (options.verbose){
1845                         printf("ActiveInterface:AvailableInterfaces:ConfiguredInterfaces:");
1846                 }
1847                 printf("\n");
1848         } else {
1849                 if (options.pnn == CTDB_BROADCAST_ALL) {
1850                         printf("Public IPs on ALL nodes\n");
1851                 } else {
1852                         printf("Public IPs on node %u\n", options.pnn);
1853                 }
1854         }
1855
1856         for (i=1;i<=ips->num;i++) {
1857                 struct ctdb_control_public_ip_info *info = NULL;
1858                 int32_t pnn;
1859                 char *aciface = NULL;
1860                 char *avifaces = NULL;
1861                 char *cifaces = NULL;
1862
1863                 if (options.pnn == CTDB_BROADCAST_ALL) {
1864                         pnn = ips->ips[ips->num-i].pnn;
1865                 } else {
1866                         pnn = options.pnn;
1867                 }
1868
1869                 if (pnn != -1) {
1870                         ret = ctdb_ctrl_get_public_ip_info(ctdb, TIMELIMIT(), pnn, ctdb,
1871                                                    &ips->ips[ips->num-i].addr, &info);
1872                 } else {
1873                         ret = -1;
1874                 }
1875
1876                 if (ret == 0) {
1877                         int j;
1878                         for (j=0; j < info->num; j++) {
1879                                 if (cifaces == NULL) {
1880                                         cifaces = talloc_strdup(info,
1881                                                                 info->ifaces[j].name);
1882                                 } else {
1883                                         cifaces = talloc_asprintf_append(cifaces,
1884                                                                          ",%s",
1885                                                                          info->ifaces[j].name);
1886                                 }
1887
1888                                 if (info->active_idx == j) {
1889                                         aciface = info->ifaces[j].name;
1890                                 }
1891
1892                                 if (info->ifaces[j].link_state == 0) {
1893                                         continue;
1894                                 }
1895
1896                                 if (avifaces == NULL) {
1897                                         avifaces = talloc_strdup(info, info->ifaces[j].name);
1898                                 } else {
1899                                         avifaces = talloc_asprintf_append(avifaces,
1900                                                                           ",%s",
1901                                                                           info->ifaces[j].name);
1902                                 }
1903                         }
1904                 }
1905
1906                 if (options.machinereadable){
1907                         printf(":%s:%d:",
1908                                 ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
1909                                 ips->ips[ips->num-i].pnn);
1910                         if (options.verbose){
1911                                 printf("%s:%s:%s:",
1912                                         aciface?aciface:"",
1913                                         avifaces?avifaces:"",
1914                                         cifaces?cifaces:"");
1915                         }
1916                         printf("\n");
1917                 } else {
1918                         if (options.verbose) {
1919                                 printf("%s node[%d] active[%s] available[%s] configured[%s]\n",
1920                                         ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
1921                                         ips->ips[ips->num-i].pnn,
1922                                         aciface?aciface:"",
1923                                         avifaces?avifaces:"",
1924                                         cifaces?cifaces:"");
1925                         } else {
1926                                 printf("%s %d\n",
1927                                         ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
1928                                         ips->ips[ips->num-i].pnn);
1929                         }
1930                 }
1931                 talloc_free(info);
1932         }
1933
1934         talloc_free(tmp_ctx);
1935         return 0;
1936 }
1937
1938 /*
1939   public ip info
1940  */
1941 static int control_ipinfo(struct ctdb_context *ctdb, int argc, const char **argv)
1942 {
1943         int i, ret;
1944         ctdb_sock_addr addr;
1945         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1946         struct ctdb_control_public_ip_info *info;
1947
1948         if (argc != 1) {
1949                 talloc_free(tmp_ctx);
1950                 usage();
1951         }
1952
1953         if (parse_ip(argv[0], NULL, 0, &addr) == 0) {
1954                 DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
1955                 return -1;
1956         }
1957
1958         /* read the public ip info from this node */
1959         ret = ctdb_ctrl_get_public_ip_info(ctdb, TIMELIMIT(), options.pnn,
1960                                            tmp_ctx, &addr, &info);
1961         if (ret != 0) {
1962                 DEBUG(DEBUG_ERR, ("Unable to get public ip[%s]info from node %u\n",
1963                                   argv[0], options.pnn));
1964                 talloc_free(tmp_ctx);
1965                 return ret;
1966         }
1967
1968         printf("Public IP[%s] info on node %u\n",
1969                ctdb_addr_to_str(&info->ip.addr),
1970                options.pnn);
1971
1972         printf("IP:%s\nCurrentNode:%d\nNumInterfaces:%u\n",
1973                ctdb_addr_to_str(&info->ip.addr),
1974                info->ip.pnn, info->num);
1975
1976         for (i=0; i<info->num; i++) {
1977                 info->ifaces[i].name[CTDB_IFACE_SIZE] = '\0';
1978
1979                 printf("Interface[%u]: Name:%s Link:%s References:%u%s\n",
1980                        i+1, info->ifaces[i].name,
1981                        info->ifaces[i].link_state?"up":"down",
1982                        (unsigned int)info->ifaces[i].references,
1983                        (i==info->active_idx)?" (active)":"");
1984         }
1985
1986         talloc_free(tmp_ctx);
1987         return 0;
1988 }
1989
1990 /*
1991   display interfaces status
1992  */
1993 static int control_ifaces(struct ctdb_context *ctdb, int argc, const char **argv)
1994 {
1995         int i, ret;
1996         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1997         struct ctdb_control_get_ifaces *ifaces;
1998
1999         /* read the public ip list from this node */
2000         ret = ctdb_ctrl_get_ifaces(ctdb, TIMELIMIT(), options.pnn,
2001                                    tmp_ctx, &ifaces);
2002         if (ret != 0) {
2003                 DEBUG(DEBUG_ERR, ("Unable to get interfaces from node %u\n",
2004                                   options.pnn));
2005                 talloc_free(tmp_ctx);
2006                 return ret;
2007         }
2008
2009         if (options.machinereadable){
2010                 printf(":Name:LinkStatus:References:\n");
2011         } else {
2012                 printf("Interfaces on node %u\n", options.pnn);
2013         }
2014
2015         for (i=0; i<ifaces->num; i++) {
2016                 if (options.machinereadable){
2017                         printf(":%s:%s:%u\n",
2018                                ifaces->ifaces[i].name,
2019                                ifaces->ifaces[i].link_state?"1":"0",
2020                                (unsigned int)ifaces->ifaces[i].references);
2021                 } else {
2022                         printf("name:%s link:%s references:%u\n",
2023                                ifaces->ifaces[i].name,
2024                                ifaces->ifaces[i].link_state?"up":"down",
2025                                (unsigned int)ifaces->ifaces[i].references);
2026                 }
2027         }
2028
2029         talloc_free(tmp_ctx);
2030         return 0;
2031 }
2032
2033
2034 /*
2035   set link status of an interface
2036  */
2037 static int control_setifacelink(struct ctdb_context *ctdb, int argc, const char **argv)
2038 {
2039         int ret;
2040         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
2041         struct ctdb_control_iface_info info;
2042
2043         ZERO_STRUCT(info);
2044
2045         if (argc != 2) {
2046                 usage();
2047         }
2048
2049         if (strlen(argv[0]) > CTDB_IFACE_SIZE) {
2050                 DEBUG(DEBUG_ERR, ("interfaces name '%s' too long\n",
2051                                   argv[0]));
2052                 talloc_free(tmp_ctx);
2053                 return -1;
2054         }
2055         strcpy(info.name, argv[0]);
2056
2057         if (strcmp(argv[1], "up") == 0) {
2058                 info.link_state = 1;
2059         } else if (strcmp(argv[1], "down") == 0) {
2060                 info.link_state = 0;
2061         } else {
2062                 DEBUG(DEBUG_ERR, ("link state invalid '%s' should be 'up' or 'down'\n",
2063                                   argv[1]));
2064                 talloc_free(tmp_ctx);
2065                 return -1;
2066         }
2067
2068         /* read the public ip list from this node */
2069         ret = ctdb_ctrl_set_iface_link(ctdb, TIMELIMIT(), options.pnn,
2070                                    tmp_ctx, &info);
2071         if (ret != 0) {
2072                 DEBUG(DEBUG_ERR, ("Unable to set link state for interfaces %s node %u\n",
2073                                   argv[0], options.pnn));
2074                 talloc_free(tmp_ctx);
2075                 return ret;
2076         }
2077
2078         talloc_free(tmp_ctx);
2079         return 0;
2080 }
2081
2082 /*
2083   display pid of a ctdb daemon
2084  */
2085 static int control_getpid(struct ctdb_context *ctdb, int argc, const char **argv)
2086 {
2087         uint32_t pid;
2088         int ret;
2089
2090         ret = ctdb_ctrl_getpid(ctdb, TIMELIMIT(), options.pnn, &pid);
2091         if (ret != 0) {
2092                 DEBUG(DEBUG_ERR, ("Unable to get daemon pid from node %u\n", options.pnn));
2093                 return ret;
2094         }
2095         printf("Pid:%d\n", pid);
2096
2097         return 0;
2098 }
2099
2100 static uint32_t ipreallocate_finished;
2101
2102 /*
2103   handler for receiving the response to ipreallocate
2104 */
2105 static void ip_reallocate_handler(struct ctdb_context *ctdb, uint64_t srvid,
2106                              TDB_DATA data, void *private_data)
2107 {
2108         ipreallocate_finished = 1;
2109 }
2110
2111 static void ctdb_every_second(struct event_context *ev, struct timed_event *te, struct timeval t, void *p)
2112 {
2113         struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
2114
2115         event_add_timed(ctdb->ev, ctdb,
2116                                 timeval_current_ofs(1, 0),
2117                                 ctdb_every_second, ctdb);
2118 }
2119
2120 /*
2121   ask the recovery daemon on the recovery master to perform a ip reallocation
2122  */
2123 static int control_ipreallocate(struct ctdb_context *ctdb, int argc, const char **argv)
2124 {
2125         int i, ret;
2126         TDB_DATA data;
2127         struct takeover_run_reply rd;
2128         uint32_t recmaster;
2129         struct ctdb_node_map *nodemap=NULL;
2130         int retries=0;
2131         struct timeval tv = timeval_current();
2132
2133         /* we need some events to trigger so we can timeout and restart
2134            the loop
2135         */
2136         event_add_timed(ctdb->ev, ctdb,
2137                                 timeval_current_ofs(1, 0),
2138                                 ctdb_every_second, ctdb);
2139
2140         rd.pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
2141         if (rd.pnn == -1) {
2142                 DEBUG(DEBUG_ERR, ("Failed to get pnn of local node\n"));
2143                 return -1;
2144         }
2145         rd.srvid = getpid();
2146
2147         /* register a message port for receiveing the reply so that we
2148            can receive the reply
2149         */
2150         ctdb_client_set_message_handler(ctdb, rd.srvid, ip_reallocate_handler, NULL);
2151
2152         data.dptr = (uint8_t *)&rd;
2153         data.dsize = sizeof(rd);
2154
2155 again:
2156         /* check that there are valid nodes available */
2157         if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap) != 0) {
2158                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2159                 return -1;
2160         }
2161         for (i=0; i<nodemap->num;i++) {
2162                 if ((nodemap->nodes[i].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) == 0) {
2163                         break;
2164                 }
2165         }
2166         if (i==nodemap->num) {
2167                 DEBUG(DEBUG_ERR,("No recmaster available, no need to wait for cluster convergence\n"));
2168                 return 0;
2169         }
2170
2171
2172         ret = ctdb_ctrl_getrecmaster(ctdb, ctdb, TIMELIMIT(), options.pnn, &recmaster);
2173         if (ret != 0) {
2174                 DEBUG(DEBUG_ERR, ("Unable to get recmaster from node %u\n", options.pnn));
2175                 return ret;
2176         }
2177
2178         /* verify the node exists */
2179         if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), recmaster, ctdb, &nodemap) != 0) {
2180                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2181                 return -1;
2182         }
2183
2184
2185         /* check tha there are nodes available that can act as a recmaster */
2186         for (i=0; i<nodemap->num; i++) {
2187                 if (nodemap->nodes[i].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
2188                         continue;
2189                 }
2190                 break;
2191         }
2192         if (i == nodemap->num) {
2193                 DEBUG(DEBUG_ERR,("No possible nodes to host addresses.\n"));
2194                 return 0;
2195         }
2196
2197         /* verify the recovery master is not STOPPED, nor BANNED */
2198         if (nodemap->nodes[recmaster].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
2199                 DEBUG(DEBUG_ERR,("No suitable recmaster found. Try again\n"));
2200                 retries++;
2201                 sleep(1);
2202                 goto again;
2203         }
2204
2205
2206         /* verify the recovery master is not STOPPED, nor BANNED */
2207         if (nodemap->nodes[recmaster].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
2208                 DEBUG(DEBUG_ERR,("No suitable recmaster found. Try again\n"));
2209                 retries++;
2210                 sleep(1);
2211                 goto again;
2212         }
2213
2214         ipreallocate_finished = 0;
2215         ret = ctdb_client_send_message(ctdb, recmaster, CTDB_SRVID_TAKEOVER_RUN, data);
2216         if (ret != 0) {
2217                 DEBUG(DEBUG_ERR,("Failed to send ip takeover run request message to %u\n", options.pnn));
2218                 return -1;
2219         }
2220
2221         tv = timeval_current();
2222         /* this loop will terminate when we have received the reply */
2223         while (timeval_elapsed(&tv) < 3.0) {
2224                 event_loop_once(ctdb->ev);
2225         }
2226         if (ipreallocate_finished == 1) {
2227                 return 0;
2228         }
2229
2230         DEBUG(DEBUG_ERR,("Timed out waiting for recmaster ipreallocate. Trying again\n"));
2231         retries++;
2232         sleep(1);
2233         goto again;
2234
2235         return 0;
2236 }
2237
2238
2239 /*
2240   disable a remote node
2241  */
2242 static int control_disable(struct ctdb_context *ctdb, int argc, const char **argv)
2243 {
2244         int ret;
2245         struct ctdb_node_map *nodemap=NULL;
2246
2247         /* check if the node is already disabled */
2248         if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
2249                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2250                 exit(10);
2251         }
2252         if (nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
2253                 DEBUG(DEBUG_ERR,("Node %d is already disabled.\n", options.pnn));
2254                 return 0;
2255         }
2256
2257         do {
2258                 ret = ctdb_ctrl_modflags(ctdb, TIMELIMIT(), options.pnn, NODE_FLAGS_PERMANENTLY_DISABLED, 0);
2259                 if (ret != 0) {
2260                         DEBUG(DEBUG_ERR, ("Unable to disable node %u\n", options.pnn));
2261                         return ret;
2262                 }
2263
2264                 sleep(1);
2265
2266                 /* read the nodemap and verify the change took effect */
2267                 if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
2268                         DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2269                         exit(10);
2270                 }
2271
2272         } while (!(nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED));
2273         ret = control_ipreallocate(ctdb, argc, argv);
2274         if (ret != 0) {
2275                 DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
2276                 return ret;
2277         }
2278
2279         return 0;
2280 }
2281
2282 /*
2283   enable a disabled remote node
2284  */
2285 static int control_enable(struct ctdb_context *ctdb, int argc, const char **argv)
2286 {
2287         int ret;
2288
2289         struct ctdb_node_map *nodemap=NULL;
2290
2291
2292         /* check if the node is already enabled */
2293         if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
2294                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2295                 exit(10);
2296         }
2297         if (!(nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED)) {
2298                 DEBUG(DEBUG_ERR,("Node %d is already enabled.\n", options.pnn));
2299                 return 0;
2300         }
2301
2302         do {
2303                 ret = ctdb_ctrl_modflags(ctdb, TIMELIMIT(), options.pnn, 0, NODE_FLAGS_PERMANENTLY_DISABLED);
2304                 if (ret != 0) {
2305                         DEBUG(DEBUG_ERR, ("Unable to enable node %u\n", options.pnn));
2306                         return ret;
2307                 }
2308
2309                 sleep(1);
2310
2311                 /* read the nodemap and verify the change took effect */
2312                 if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
2313                         DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2314                         exit(10);
2315                 }
2316
2317         } while (nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED);
2318
2319         ret = control_ipreallocate(ctdb, argc, argv);
2320         if (ret != 0) {
2321                 DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
2322                 return ret;
2323         }
2324
2325         return 0;
2326 }
2327
2328 /*
2329   stop a remote node
2330  */
2331 static int control_stop(struct ctdb_context *ctdb, int argc, const char **argv)
2332 {
2333         int ret;
2334         struct ctdb_node_map *nodemap=NULL;
2335
2336         do {
2337                 ret = ctdb_ctrl_stop_node(ctdb, TIMELIMIT(), options.pnn);
2338                 if (ret != 0) {
2339                         DEBUG(DEBUG_ERR, ("Unable to stop node %u   try again\n", options.pnn));
2340                 }
2341
2342                 sleep(1);
2343
2344                 /* read the nodemap and verify the change took effect */
2345                 if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
2346                         DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2347                         exit(10);
2348                 }
2349
2350         } while (!(nodemap->nodes[options.pnn].flags & NODE_FLAGS_STOPPED));
2351         ret = control_ipreallocate(ctdb, argc, argv);
2352         if (ret != 0) {
2353                 DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
2354                 return ret;
2355         }
2356
2357         return 0;
2358 }
2359
2360 /*
2361   restart a stopped remote node
2362  */
2363 static int control_continue(struct ctdb_context *ctdb, int argc, const char **argv)
2364 {
2365         int ret;
2366
2367         struct ctdb_node_map *nodemap=NULL;
2368
2369         do {
2370                 ret = ctdb_ctrl_continue_node(ctdb, TIMELIMIT(), options.pnn);
2371                 if (ret != 0) {
2372                         DEBUG(DEBUG_ERR, ("Unable to continue node %u\n", options.pnn));
2373                         return ret;
2374                 }
2375
2376                 sleep(1);
2377
2378                 /* read the nodemap and verify the change took effect */
2379                 if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
2380                         DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2381                         exit(10);
2382                 }
2383
2384         } while (nodemap->nodes[options.pnn].flags & NODE_FLAGS_STOPPED);
2385         ret = control_ipreallocate(ctdb, argc, argv);
2386         if (ret != 0) {
2387                 DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
2388                 return ret;
2389         }
2390
2391         return 0;
2392 }
2393
2394 static uint32_t get_generation(struct ctdb_context *ctdb)
2395 {
2396         struct ctdb_vnn_map *vnnmap=NULL;
2397         int ret;
2398
2399         /* wait until the recmaster is not in recovery mode */
2400         while (1) {
2401                 uint32_t recmode, recmaster;
2402
2403                 if (vnnmap != NULL) {
2404                         talloc_free(vnnmap);
2405                         vnnmap = NULL;
2406                 }
2407
2408                 /* get the recmaster */
2409                 ret = ctdb_ctrl_getrecmaster(ctdb, ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, &recmaster);
2410                 if (ret != 0) {
2411                         DEBUG(DEBUG_ERR, ("Unable to get recmaster from node %u\n", options.pnn));
2412                         exit(10);
2413                 }
2414
2415                 /* get recovery mode */
2416                 ret = ctdb_ctrl_getrecmode(ctdb, ctdb, TIMELIMIT(), recmaster, &recmode);
2417                 if (ret != 0) {
2418                         DEBUG(DEBUG_ERR, ("Unable to get recmode from node %u\n", options.pnn));
2419                         exit(10);
2420                 }
2421
2422                 /* get the current generation number */
2423                 ret = ctdb_ctrl_getvnnmap(ctdb, TIMELIMIT(), recmaster, ctdb, &vnnmap);
2424                 if (ret != 0) {
2425                         DEBUG(DEBUG_ERR, ("Unable to get vnnmap from recmaster (%u)\n", recmaster));
2426                         exit(10);
2427                 }
2428
2429                 if ((recmode == CTDB_RECOVERY_NORMAL)
2430                 &&  (vnnmap->generation != 1)){
2431                         return vnnmap->generation;
2432                 }
2433                 sleep(1);
2434         }
2435 }
2436
2437 /*
2438   ban a node from the cluster
2439  */
2440 static int control_ban(struct ctdb_context *ctdb, int argc, const char **argv)
2441 {
2442         int ret;
2443         struct ctdb_node_map *nodemap=NULL;
2444         struct ctdb_ban_time bantime;
2445
2446         if (argc < 1) {
2447                 usage();
2448         }
2449
2450         /* verify the node exists */
2451         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap);
2452         if (ret != 0) {
2453                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2454                 return ret;
2455         }
2456
2457         if (nodemap->nodes[options.pnn].flags & NODE_FLAGS_BANNED) {
2458                 DEBUG(DEBUG_ERR,("Node %u is already banned.\n", options.pnn));
2459                 return -1;
2460         }
2461
2462         bantime.pnn  = options.pnn;
2463         bantime.time = strtoul(argv[0], NULL, 0);
2464
2465         ret = ctdb_ctrl_set_ban(ctdb, TIMELIMIT(), options.pnn, &bantime);
2466         if (ret != 0) {
2467                 DEBUG(DEBUG_ERR,("Banning node %d for %d seconds failed.\n", bantime.pnn, bantime.time));
2468                 return -1;
2469         }
2470
2471         ret = control_ipreallocate(ctdb, argc, argv);
2472         if (ret != 0) {
2473                 DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
2474                 return ret;
2475         }
2476
2477         return 0;
2478 }
2479
2480
2481 /*
2482   unban a node from the cluster
2483  */
2484 static int control_unban(struct ctdb_context *ctdb, int argc, const char **argv)
2485 {
2486         int ret;
2487         struct ctdb_node_map *nodemap=NULL;
2488         struct ctdb_ban_time bantime;
2489
2490         /* verify the node exists */
2491         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap);
2492         if (ret != 0) {
2493                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2494                 return ret;
2495         }
2496
2497         if (!(nodemap->nodes[options.pnn].flags & NODE_FLAGS_BANNED)) {
2498                 DEBUG(DEBUG_ERR,("Node %u is not banned.\n", options.pnn));
2499                 return -1;
2500         }
2501
2502         bantime.pnn  = options.pnn;
2503         bantime.time = 0;
2504
2505         ret = ctdb_ctrl_set_ban(ctdb, TIMELIMIT(), options.pnn, &bantime);
2506         if (ret != 0) {
2507                 DEBUG(DEBUG_ERR,("Unbanning node %d failed.\n", bantime.pnn));
2508                 return -1;
2509         }
2510
2511         ret = control_ipreallocate(ctdb, argc, argv);
2512         if (ret != 0) {
2513                 DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
2514                 return ret;
2515         }
2516
2517         return 0;
2518 }
2519
2520
2521 /*
2522   show ban information for a node
2523  */
2524 static int control_showban(struct ctdb_context *ctdb, int argc, const char **argv)
2525 {
2526         int ret;
2527         struct ctdb_node_map *nodemap=NULL;
2528         struct ctdb_ban_time *bantime;
2529
2530         /* verify the node exists */
2531         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap);
2532         if (ret != 0) {
2533                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
2534                 return ret;
2535         }
2536
2537         ret = ctdb_ctrl_get_ban(ctdb, TIMELIMIT(), options.pnn, ctdb, &bantime);
2538         if (ret != 0) {
2539                 DEBUG(DEBUG_ERR,("Showing ban info for node %d failed.\n", options.pnn));
2540                 return -1;
2541         }
2542
2543         if (bantime->time == 0) {
2544                 printf("Node %u is not banned\n", bantime->pnn);
2545         } else {
2546                 printf("Node %u is banned banned for %d seconds\n", bantime->pnn, bantime->time);
2547         }
2548
2549         return 0;
2550 }
2551
2552 /*
2553   shutdown a daemon
2554  */
2555 static int control_shutdown(struct ctdb_context *ctdb, int argc, const char **argv)
2556 {
2557         int ret;
2558
2559         ret = ctdb_ctrl_shutdown(ctdb, TIMELIMIT(), options.pnn);
2560         if (ret != 0) {
2561                 DEBUG(DEBUG_ERR, ("Unable to shutdown node %u\n", options.pnn));
2562                 return ret;
2563         }
2564
2565         return 0;
2566 }
2567
2568 /*
2569   trigger a recovery
2570  */
2571 static int control_recover(struct ctdb_context *ctdb, int argc, const char **argv)
2572 {
2573         int ret;
2574         uint32_t generation, next_generation;
2575
2576         /* record the current generation number */
2577         generation = get_generation(ctdb);
2578
2579         ret = ctdb_ctrl_freeze_priority(ctdb, TIMELIMIT(), options.pnn, 1);
2580         if (ret != 0) {
2581                 DEBUG(DEBUG_ERR, ("Unable to freeze node\n"));
2582                 return ret;
2583         }
2584
2585         ret = ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
2586         if (ret != 0) {
2587                 DEBUG(DEBUG_ERR, ("Unable to set recovery mode\n"));
2588                 return ret;
2589         }
2590
2591         /* wait until we are in a new generation */
2592         while (1) {
2593                 next_generation = get_generation(ctdb);
2594                 if (next_generation != generation) {
2595                         return 0;
2596                 }
2597                 sleep(1);
2598         }
2599
2600         return 0;
2601 }
2602
2603
2604 /*
2605   display monitoring mode of a remote node
2606  */
2607 static int control_getmonmode(struct ctdb_context *ctdb, int argc, const char **argv)
2608 {
2609         uint32_t monmode;
2610         int ret;
2611
2612         ret = ctdb_ctrl_getmonmode(ctdb, TIMELIMIT(), options.pnn, &monmode);
2613         if (ret != 0) {
2614                 DEBUG(DEBUG_ERR, ("Unable to get monmode from node %u\n", options.pnn));
2615                 return ret;
2616         }
2617         if (!options.machinereadable){
2618                 printf("Monitoring mode:%s (%d)\n",monmode==CTDB_MONITORING_ACTIVE?"ACTIVE":"DISABLED",monmode);
2619         } else {
2620                 printf(":mode:\n");
2621                 printf(":%d:\n",monmode);
2622         }
2623         return 0;
2624 }
2625
2626
2627 /*
2628   display capabilities of a remote node
2629  */
2630 static int control_getcapabilities(struct ctdb_context *ctdb, int argc, const char **argv)
2631 {
2632         uint32_t capabilities;
2633         int ret;
2634
2635         ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), options.pnn, &capabilities);
2636         if (ret != 0) {
2637                 DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", options.pnn));
2638                 return ret;
2639         }
2640
2641         if (!options.machinereadable){
2642                 printf("RECMASTER: %s\n", (capabilities&CTDB_CAP_RECMASTER)?"YES":"NO");
2643                 printf("LMASTER: %s\n", (capabilities&CTDB_CAP_LMASTER)?"YES":"NO");
2644                 printf("LVS: %s\n", (capabilities&CTDB_CAP_LVS)?"YES":"NO");
2645                 printf("NATGW: %s\n", (capabilities&CTDB_CAP_NATGW)?"YES":"NO");
2646         } else {
2647                 printf(":RECMASTER:LMASTER:LVS:NATGW:\n");
2648                 printf(":%d:%d:%d:%d:\n",
2649                         !!(capabilities&CTDB_CAP_RECMASTER),
2650                         !!(capabilities&CTDB_CAP_LMASTER),
2651                         !!(capabilities&CTDB_CAP_LVS),
2652                         !!(capabilities&CTDB_CAP_NATGW));
2653         }
2654         return 0;
2655 }
2656
2657 /*
2658   display lvs configuration
2659  */
2660 static int control_lvs(struct ctdb_context *ctdb, int argc, const char **argv)
2661 {
2662         uint32_t *capabilities;
2663         struct ctdb_node_map *nodemap=NULL;
2664         int i, ret;
2665         int healthy_count = 0;
2666
2667         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap);
2668         if (ret != 0) {
2669                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
2670                 return ret;
2671         }
2672
2673         capabilities = talloc_array(ctdb, uint32_t, nodemap->num);
2674         CTDB_NO_MEMORY(ctdb, capabilities);
2675
2676         /* collect capabilities for all connected nodes */
2677         for (i=0; i<nodemap->num; i++) {
2678                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
2679                         continue;
2680                 }
2681                 if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
2682                         continue;
2683                 }
2684
2685                 ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), i, &capabilities[i]);
2686                 if (ret != 0) {
2687                         DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", i));
2688                         return ret;
2689                 }
2690
2691                 if (!(capabilities[i] & CTDB_CAP_LVS)) {
2692                         continue;
2693                 }
2694
2695                 if (!(nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY)) {
2696                         healthy_count++;
2697                 }
2698         }
2699
2700         /* Print all LVS nodes */
2701         for (i=0; i<nodemap->num; i++) {
2702                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
2703                         continue;
2704                 }
2705                 if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
2706                         continue;
2707                 }
2708                 if (!(capabilities[i] & CTDB_CAP_LVS)) {
2709                         continue;
2710                 }
2711
2712                 if (healthy_count != 0) {
2713                         if (nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY) {
2714                                 continue;
2715                         }
2716                 }
2717
2718                 printf("%d:%s\n", i,
2719                         ctdb_addr_to_str(&nodemap->nodes[i].addr));
2720         }
2721
2722         return 0;
2723 }
2724
2725 /*
2726   display who is the lvs master
2727  */
2728 static int control_lvsmaster(struct ctdb_context *ctdb, int argc, const char **argv)
2729 {
2730         uint32_t *capabilities;
2731         struct ctdb_node_map *nodemap=NULL;
2732         int i, ret;
2733         int healthy_count = 0;
2734
2735         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap);
2736         if (ret != 0) {
2737                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
2738                 return ret;
2739         }
2740
2741         capabilities = talloc_array(ctdb, uint32_t, nodemap->num);
2742         CTDB_NO_MEMORY(ctdb, capabilities);
2743
2744         /* collect capabilities for all connected nodes */
2745         for (i=0; i<nodemap->num; i++) {
2746                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
2747                         continue;
2748                 }
2749                 if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
2750                         continue;
2751                 }
2752
2753                 ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), i, &capabilities[i]);
2754                 if (ret != 0) {
2755                         DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", i));
2756                         return ret;
2757                 }
2758
2759                 if (!(capabilities[i] & CTDB_CAP_LVS)) {
2760                         continue;
2761                 }
2762
2763                 if (!(nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY)) {
2764                         healthy_count++;
2765                 }
2766         }
2767
2768         /* find and show the lvsmaster */
2769         for (i=0; i<nodemap->num; i++) {
2770                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
2771                         continue;
2772                 }
2773                 if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
2774                         continue;
2775                 }
2776                 if (!(capabilities[i] & CTDB_CAP_LVS)) {
2777                         continue;
2778                 }
2779
2780                 if (healthy_count != 0) {
2781                         if (nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY) {
2782                                 continue;
2783                         }
2784                 }
2785
2786                 if (options.machinereadable){
2787                         printf("%d\n", i);
2788                 } else {
2789                         printf("Node %d is LVS master\n", i);
2790                 }
2791                 return 0;
2792         }
2793
2794         printf("There is no LVS master\n");
2795         return -1;
2796 }
2797
2798 /*
2799   disable monitoring on a  node
2800  */
2801 static int control_disable_monmode(struct ctdb_context *ctdb, int argc, const char **argv)
2802 {
2803
2804         int ret;
2805
2806         ret = ctdb_ctrl_disable_monmode(ctdb, TIMELIMIT(), options.pnn);
2807         if (ret != 0) {
2808                 DEBUG(DEBUG_ERR, ("Unable to disable monmode on node %u\n", options.pnn));
2809                 return ret;
2810         }
2811         printf("Monitoring mode:%s\n","DISABLED");
2812
2813         return 0;
2814 }
2815
2816 /*
2817   enable monitoring on a  node
2818  */
2819 static int control_enable_monmode(struct ctdb_context *ctdb, int argc, const char **argv)
2820 {
2821
2822         int ret;
2823
2824         ret = ctdb_ctrl_enable_monmode(ctdb, TIMELIMIT(), options.pnn);
2825         if (ret != 0) {
2826                 DEBUG(DEBUG_ERR, ("Unable to enable monmode on node %u\n", options.pnn));
2827                 return ret;
2828         }
2829         printf("Monitoring mode:%s\n","ACTIVE");
2830
2831         return 0;
2832 }
2833
2834 /*
2835   display remote list of keys/data for a db
2836  */
2837 static int control_catdb(struct ctdb_context *ctdb, int argc, const char **argv)
2838 {
2839         const char *db_name;
2840         struct ctdb_db_context *ctdb_db;
2841         int ret;
2842
2843         if (argc < 1) {
2844                 usage();
2845         }
2846
2847         db_name = argv[0];
2848
2849
2850         if (db_exists(ctdb, db_name)) {
2851                 DEBUG(DEBUG_ERR,("Database '%s' does not exist\n", db_name));
2852                 return -1;
2853         }
2854
2855         ctdb_db = ctdb_attach(ctdb, db_name, false, 0);
2856
2857         if (ctdb_db == NULL) {
2858                 DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
2859                 return -1;
2860         }
2861
2862         /* traverse and dump the cluster tdb */
2863         ret = ctdb_dump_db(ctdb_db, stdout);
2864         if (ret == -1) {
2865                 DEBUG(DEBUG_ERR, ("Unable to dump database\n"));
2866                 DEBUG(DEBUG_ERR, ("Maybe try 'ctdb getdbstatus %s'"
2867                                   " and 'ctdb getvar AllowUnhealthyDBRead'\n",
2868                                   db_name));
2869                 return -1;
2870         }
2871         talloc_free(ctdb_db);
2872
2873         printf("Dumped %d records\n", ret);
2874         return 0;
2875 }
2876
2877
2878 /*
2879   fetch a record from a persistent database
2880  */
2881 static int control_pfetch(struct ctdb_context *ctdb, int argc, const char **argv)
2882 {
2883         const char *db_name;
2884         struct ctdb_db_context *ctdb_db;
2885         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
2886         struct ctdb_transaction_handle *h;
2887         TDB_DATA key, data;
2888         int fd, ret;
2889
2890         if (argc < 2) {
2891                 talloc_free(tmp_ctx);
2892                 usage();
2893         }
2894
2895         db_name = argv[0];
2896
2897
2898         if (db_exists(ctdb, db_name)) {
2899                 DEBUG(DEBUG_ERR,("Database '%s' does not exist\n", db_name));
2900                 talloc_free(tmp_ctx);
2901                 return -1;
2902         }
2903
2904         ctdb_db = ctdb_attach(ctdb, db_name, true, 0);
2905
2906         if (ctdb_db == NULL) {
2907                 DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
2908                 talloc_free(tmp_ctx);
2909                 return -1;
2910         }
2911
2912         h = ctdb_transaction_start(ctdb_db, tmp_ctx);
2913         if (h == NULL) {
2914                 DEBUG(DEBUG_ERR,("Failed to start transaction on database %s\n", db_name));
2915                 talloc_free(tmp_ctx);
2916                 return -1;
2917         }
2918
2919         key.dptr  = discard_const(argv[1]);
2920         key.dsize = strlen(argv[1]);
2921         ret = ctdb_transaction_fetch(h, tmp_ctx, key, &data);
2922         if (ret != 0) {
2923                 DEBUG(DEBUG_ERR,("Failed to fetch record\n"));
2924                 talloc_free(tmp_ctx);
2925                 return -1;
2926         }
2927
2928         if (data.dsize == 0 || data.dptr == NULL) {
2929                 DEBUG(DEBUG_ERR,("Record is empty\n"));
2930                 talloc_free(tmp_ctx);
2931                 return -1;
2932         }
2933
2934         if (argc == 3) {
2935           fd = open(argv[2], O_WRONLY|O_CREAT|O_TRUNC, 0600);
2936                 if (fd == -1) {
2937                         DEBUG(DEBUG_ERR,("Failed to open output file %s\n", argv[2]));
2938                         talloc_free(tmp_ctx);
2939                         return -1;
2940                 }
2941                 write(fd, data.dptr, data.dsize);
2942                 close(fd);
2943         } else {
2944                 write(1, data.dptr, data.dsize);
2945         }
2946
2947         /* abort the transaction */
2948         talloc_free(h);
2949
2950
2951         talloc_free(tmp_ctx);
2952         return 0;
2953 }
2954
2955 /*
2956   fetch a record from a tdb-file
2957  */
2958 static int control_tfetch(struct ctdb_context *ctdb, int argc, const char **argv)
2959 {
2960         const char *tdb_file;
2961         TDB_CONTEXT *tdb;
2962         TDB_DATA key, data;
2963         int fd;
2964
2965         if (argc < 2) {
2966                 usage();
2967         }
2968
2969         tdb_file = argv[0];
2970
2971         tdb = tdb_open(tdb_file, 0, 0, O_RDONLY, 0);
2972         if (tdb == NULL) {
2973                 DEBUG(DEBUG_ERR,("Failed to open TDB file %s\n", tdb_file));
2974                 return -1;
2975         }
2976
2977         key.dptr  = discard_const(argv[1]);
2978         key.dsize = strlen(argv[1]);
2979         data = tdb_fetch(tdb, key);
2980         if (data.dptr == NULL || data.dsize < sizeof(struct ctdb_ltdb_header)) {
2981                 DEBUG(DEBUG_ERR,("Failed to read record %s from tdb %s\n", argv[1], tdb_file));
2982                 tdb_close(tdb);
2983                 return -1;
2984         }
2985
2986         tdb_close(tdb);
2987
2988         if (argc == 3) {
2989           fd = open(argv[2], O_WRONLY|O_CREAT|O_TRUNC, 0600);
2990                 if (fd == -1) {
2991                         DEBUG(DEBUG_ERR,("Failed to open output file %s\n", argv[2]));
2992                         return -1;
2993                 }
2994                 write(fd, data.dptr+sizeof(struct ctdb_ltdb_header), data.dsize-sizeof(struct ctdb_ltdb_header));
2995                 close(fd);
2996         } else {
2997                 write(1, data.dptr+sizeof(struct ctdb_ltdb_header), data.dsize-sizeof(struct ctdb_ltdb_header));
2998         }
2999
3000         return 0;
3001 }
3002
3003 /*
3004   write a record to a persistent database
3005  */
3006 static int control_pstore(struct ctdb_context *ctdb, int argc, const char **argv)
3007 {
3008         const char *db_name;
3009         struct ctdb_db_context *ctdb_db;
3010         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
3011         struct ctdb_transaction_handle *h;
3012         struct stat st;
3013         TDB_DATA key, data;
3014         int fd, ret;
3015
3016         if (argc < 3) {
3017                 talloc_free(tmp_ctx);
3018                 usage();
3019         }
3020
3021         fd = open(argv[2], O_RDONLY);
3022         if (fd == -1) {
3023                 DEBUG(DEBUG_ERR,("Failed to open file containing record data : %s  %s\n", argv[2], strerror(errno)));
3024                 talloc_free(tmp_ctx);
3025                 return -1;
3026         }
3027
3028         ret = fstat(fd, &st);
3029         if (ret == -1) {
3030                 DEBUG(DEBUG_ERR,("fstat of file %s failed: %s\n", argv[2], strerror(errno)));
3031                 close(fd);
3032                 talloc_free(tmp_ctx);
3033                 return -1;
3034         }
3035
3036         if (!S_ISREG(st.st_mode)) {
3037                 DEBUG(DEBUG_ERR,("Not a regular file %s\n", argv[2]));
3038                 close(fd);
3039                 talloc_free(tmp_ctx);
3040                 return -1;
3041         }
3042
3043         data.dsize = st.st_size;
3044         if (data.dsize == 0) {
3045                 data.dptr  = NULL;
3046         } else {
3047                 data.dptr = talloc_size(tmp_ctx, data.dsize);
3048                 if (data.dptr == NULL) {
3049                         DEBUG(DEBUG_ERR,("Failed to talloc %d of memory to store record data\n", (int)data.dsize));
3050                         close(fd);
3051                         talloc_free(tmp_ctx);
3052                         return -1;
3053                 }
3054                 ret = read(fd, data.dptr, data.dsize);
3055                 if (ret != data.dsize) {
3056                         DEBUG(DEBUG_ERR,("Failed to read %d bytes of record data\n", (int)data.dsize));
3057                         close(fd);
3058                         talloc_free(tmp_ctx);
3059                         return -1;
3060                 }
3061         }
3062         close(fd);
3063
3064
3065         db_name = argv[0];
3066
3067         ctdb_db = ctdb_attach(ctdb, db_name, true, 0);
3068
3069         if (ctdb_db == NULL) {
3070                 DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
3071                 talloc_free(tmp_ctx);
3072                 return -1;
3073         }
3074
3075         h = ctdb_transaction_start(ctdb_db, tmp_ctx);
3076         if (h == NULL) {
3077                 DEBUG(DEBUG_ERR,("Failed to start transaction on database %s\n", db_name));
3078                 talloc_free(tmp_ctx);
3079                 return -1;
3080         }
3081
3082         key.dptr  = discard_const(argv[1]);
3083         key.dsize = strlen(argv[1]);
3084         ret = ctdb_transaction_store(h, key, data);
3085         if (ret != 0) {
3086                 DEBUG(DEBUG_ERR,("Failed to store record\n"));
3087                 talloc_free(tmp_ctx);
3088                 return -1;
3089         }
3090
3091         ret = ctdb_transaction_commit(h);
3092         if (ret != 0) {
3093                 DEBUG(DEBUG_ERR,("Failed to commit transaction\n"));
3094                 talloc_free(tmp_ctx);
3095                 return -1;
3096         }
3097
3098
3099         talloc_free(tmp_ctx);
3100         return 0;
3101 }
3102
3103 static void log_handler(struct ctdb_context *ctdb, uint64_t srvid,
3104                              TDB_DATA data, void *private_data)
3105 {
3106         DEBUG(DEBUG_ERR,("Log data received\n"));
3107         if (data.dsize > 0) {
3108                 printf("%s", data.dptr);
3109         }
3110
3111         exit(0);
3112 }
3113
3114 /*
3115   display a list of log messages from the in memory ringbuffer
3116  */
3117 static int control_getlog(struct ctdb_context *ctdb, int argc, const char **argv)
3118 {
3119         int ret;
3120         int32_t res;
3121         struct ctdb_get_log_addr log_addr;
3122         TDB_DATA data;
3123         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
3124         char *errmsg;
3125         struct timeval tv;
3126
3127         if (argc != 1) {
3128                 DEBUG(DEBUG_ERR,("Invalid arguments\n"));
3129                 talloc_free(tmp_ctx);
3130                 return -1;
3131         }
3132
3133         log_addr.pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
3134         log_addr.srvid = getpid();
3135         if (isalpha(argv[0][0]) || argv[0][0] == '-') {
3136                 log_addr.level = get_debug_by_desc(argv[0]);
3137         } else {
3138                 log_addr.level = strtol(argv[0], NULL, 0);
3139         }
3140
3141
3142         data.dptr = (unsigned char *)&log_addr;
3143         data.dsize = sizeof(log_addr);
3144
3145         DEBUG(DEBUG_ERR, ("Pulling logs from node %u\n", options.pnn));
3146
3147         ctdb_client_set_message_handler(ctdb, log_addr.srvid, log_handler, NULL);
3148         sleep(1);
3149
3150         DEBUG(DEBUG_ERR,("Listen for response on %d\n", (int)log_addr.srvid));
3151
3152         ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_GET_LOG,
3153                            0, data, tmp_ctx, NULL, &res, NULL, &errmsg);
3154         if (ret != 0 || res != 0) {
3155                 DEBUG(DEBUG_ERR,("Failed to get logs - %s\n", errmsg));
3156                 talloc_free(tmp_ctx);
3157                 return -1;
3158         }
3159
3160
3161         tv = timeval_current();
3162         /* this loop will terminate when we have received the reply */
3163         while (timeval_elapsed(&tv) < 3.0) {
3164                 event_loop_once(ctdb->ev);
3165         }
3166
3167         DEBUG(DEBUG_INFO,("Timed out waiting for log data.\n"));
3168
3169         talloc_free(tmp_ctx);
3170         return 0;
3171 }
3172
3173 /*
3174   clear the in memory log area
3175  */
3176 static int control_clearlog(struct ctdb_context *ctdb, int argc, const char **argv)
3177 {
3178         int ret;
3179         int32_t res;
3180         char *errmsg;
3181         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
3182
3183         ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_CLEAR_LOG,
3184                            0, tdb_null, tmp_ctx, NULL, &res, NULL, &errmsg);
3185         if (ret != 0 || res != 0) {
3186                 DEBUG(DEBUG_ERR,("Failed to clear logs\n"));
3187                 talloc_free(tmp_ctx);
3188                 return -1;
3189         }
3190
3191         talloc_free(tmp_ctx);
3192         return 0;
3193 }
3194
3195
3196
3197 /*
3198   display a list of the databases on a remote ctdb
3199  */
3200 static int control_getdbmap(struct ctdb_context *ctdb, int argc, const char **argv)
3201 {
3202         int i, ret;
3203         struct ctdb_dbid_map *dbmap=NULL;
3204
3205         ret = ctdb_ctrl_getdbmap(ctdb, TIMELIMIT(), options.pnn, ctdb, &dbmap);
3206         if (ret != 0) {
3207                 DEBUG(DEBUG_ERR, ("Unable to get dbids from node %u\n", options.pnn));
3208                 return ret;
3209         }
3210
3211         if(options.machinereadable){
3212                 printf(":ID:Name:Path:Persistent:Unhealthy:\n");
3213                 for(i=0;i<dbmap->num;i++){
3214                         const char *path;
3215                         const char *name;
3216                         const char *health;
3217                         bool persistent;
3218
3219                         ctdb_ctrl_getdbpath(ctdb, TIMELIMIT(), options.pnn,
3220                                             dbmap->dbs[i].dbid, ctdb, &path);
3221                         ctdb_ctrl_getdbname(ctdb, TIMELIMIT(), options.pnn,
3222                                             dbmap->dbs[i].dbid, ctdb, &name);
3223                         ctdb_ctrl_getdbhealth(ctdb, TIMELIMIT(), options.pnn,
3224                                               dbmap->dbs[i].dbid, ctdb, &health);
3225                         persistent = dbmap->dbs[i].persistent;
3226                         printf(":0x%08X:%s:%s:%d:%d:\n",
3227                                dbmap->dbs[i].dbid, name, path,
3228                                !!(persistent), !!(health));
3229                 }
3230                 return 0;
3231         }
3232
3233         printf("Number of databases:%d\n", dbmap->num);
3234         for(i=0;i<dbmap->num;i++){
3235                 const char *path;
3236                 const char *name;
3237                 const char *health;
3238                 bool persistent;
3239
3240                 ctdb_ctrl_getdbpath(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, ctdb, &path);
3241                 ctdb_ctrl_getdbname(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, ctdb, &name);
3242                 ctdb_ctrl_getdbhealth(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, ctdb, &health);
3243                 persistent = dbmap->dbs[i].persistent;
3244                 printf("dbid:0x%08x name:%s path:%s%s%s\n",
3245                        dbmap->dbs[i].dbid, name, path,
3246                        persistent?" PERSISTENT":"",
3247                        health?" UNHEALTHY":"");
3248         }
3249
3250         return 0;
3251 }
3252
3253 /*
3254   display the status of a database on a remote ctdb
3255  */
3256 static int control_getdbstatus(struct ctdb_context *ctdb, int argc, const char **argv)
3257 {
3258         int i, ret;
3259         struct ctdb_dbid_map *dbmap=NULL;
3260         const char *db_name;
3261
3262         if (argc < 1) {
3263                 usage();
3264         }
3265
3266         db_name = argv[0];
3267
3268         ret = ctdb_ctrl_getdbmap(ctdb, TIMELIMIT(), options.pnn, ctdb, &dbmap);
3269         if (ret != 0) {
3270                 DEBUG(DEBUG_ERR, ("Unable to get dbids from node %u\n", options.pnn));
3271                 return ret;
3272         }
3273
3274         for(i=0;i<dbmap->num;i++){
3275                 const char *path;
3276                 const char *name;
3277                 const char *health;
3278                 bool persistent;
3279
3280                 ctdb_ctrl_getdbname(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, ctdb, &name);
3281                 if (strcmp(name, db_name) != 0) {
3282                         continue;
3283                 }
3284
3285                 ctdb_ctrl_getdbpath(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, ctdb, &path);
3286                 ctdb_ctrl_getdbhealth(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, ctdb, &health);
3287                 persistent = dbmap->dbs[i].persistent;
3288                 printf("dbid: 0x%08x\nname: %s\npath: %s\nPERSISTENT: %s\nHEALTH: %s\n",
3289                        dbmap->dbs[i].dbid, name, path,
3290                        persistent?"yes":"no",
3291                        health?health:"OK");
3292                 return 0;
3293         }
3294
3295         DEBUG(DEBUG_ERR, ("db %s doesn't exist on node %u\n", db_name, options.pnn));
3296         return 0;
3297 }
3298
3299 /*
3300   check if the local node is recmaster or not
3301   it will return 1 if this node is the recmaster and 0 if it is not
3302   or if the local ctdb daemon could not be contacted
3303  */
3304 static int control_isnotrecmaster(struct ctdb_context *ctdb, int argc, const char **argv)
3305 {
3306         uint32_t mypnn, recmaster;
3307         int ret;
3308
3309         mypnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);
3310         if (mypnn == -1) {
3311                 printf("Failed to get pnn of node\n");
3312                 return 1;
3313         }
3314
3315         ret = ctdb_ctrl_getrecmaster(ctdb, ctdb, TIMELIMIT(), options.pnn, &recmaster);
3316         if (ret != 0) {
3317                 printf("Failed to get the recmaster\n");
3318                 return 1;
3319         }
3320
3321         if (recmaster != mypnn) {
3322                 printf("this node is not the recmaster\n");
3323                 return 1;
3324         }
3325
3326         printf("this node is the recmaster\n");
3327         return 0;
3328 }
3329
3330 /*
3331   ping a node
3332  */
3333 static int control_ping(struct ctdb_context *ctdb, int argc, const char **argv)
3334 {
3335         int ret;
3336         struct timeval tv = timeval_current();
3337         ret = ctdb_ctrl_ping(ctdb, options.pnn);
3338         if (ret == -1) {
3339                 printf("Unable to get ping response from node %u\n", options.pnn);
3340                 return -1;
3341         } else {
3342                 printf("response from %u time=%.6f sec  (%d clients)\n",
3343                        options.pnn, timeval_elapsed(&tv), ret);
3344         }
3345         return 0;
3346 }
3347
3348
3349 /*
3350   get a tunable
3351  */
3352 static int control_getvar(struct ctdb_context *ctdb, int argc, const char **argv)
3353 {
3354         const char *name;
3355         uint32_t value;
3356         int ret;
3357
3358         if (argc < 1) {
3359                 usage();
3360         }
3361
3362         name = argv[0];
3363         ret = ctdb_ctrl_get_tunable(ctdb, TIMELIMIT(), options.pnn, name, &value);
3364         if (ret == -1) {
3365                 DEBUG(DEBUG_ERR, ("Unable to get tunable variable '%s'\n", name));
3366                 return -1;
3367         }
3368
3369         printf("%-19s = %u\n", name, value);
3370         return 0;
3371 }
3372
3373 /*
3374   set a tunable
3375  */
3376 static int control_setvar(struct ctdb_context *ctdb, int argc, const char **argv)
3377 {
3378         const char *name;
3379         uint32_t value;
3380         int ret;
3381
3382         if (argc < 2) {
3383                 usage();
3384         }
3385
3386         name = argv[0];
3387         value = strtoul(argv[1], NULL, 0);
3388
3389         ret = ctdb_ctrl_set_tunable(ctdb, TIMELIMIT(), options.pnn, name, value);
3390         if (ret == -1) {
3391                 DEBUG(DEBUG_ERR, ("Unable to set tunable variable '%s'\n", name));
3392                 return -1;
3393         }
3394         return 0;
3395 }
3396
3397 /*
3398   list all tunables
3399  */
3400 static int control_listvars(struct ctdb_context *ctdb, int argc, const char **argv)
3401 {
3402         uint32_t count;
3403         const char **list;
3404         int ret, i;
3405
3406         ret = ctdb_ctrl_list_tunables(ctdb, TIMELIMIT(), options.pnn, ctdb, &list, &count);
3407         if (ret == -1) {
3408                 DEBUG(DEBUG_ERR, ("Unable to list tunable variables\n"));
3409                 return -1;
3410         }
3411
3412         for (i=0;i<count;i++) {
3413                 control_getvar(ctdb, 1, &list[i]);
3414         }
3415
3416         talloc_free(list);
3417
3418         return 0;
3419 }
3420
3421 /*
3422   display debug level on a node
3423  */
3424 static int control_getdebug(struct ctdb_context *ctdb, int argc, const char **argv)
3425 {
3426         int ret;
3427         int32_t level;
3428
3429         ret = ctdb_ctrl_get_debuglevel(ctdb, options.pnn, &level);
3430         if (ret != 0) {
3431                 DEBUG(DEBUG_ERR, ("Unable to get debuglevel response from node %u\n", options.pnn));
3432                 return ret;
3433         } else {
3434                 if (options.machinereadable){
3435                         printf(":Name:Level:\n");
3436                         printf(":%s:%d:\n",get_debug_by_level(level),level);
3437                 } else {
3438                         printf("Node %u is at debug level %s (%d)\n", options.pnn, get_debug_by_level(level), level);
3439                 }
3440         }
3441         return 0;
3442 }
3443
3444 /*
3445   display reclock file of a node
3446  */
3447 static int control_getreclock(struct ctdb_context *ctdb, int argc, const char **argv)
3448 {
3449         int ret;
3450         const char *reclock;
3451
3452         ret = ctdb_ctrl_getreclock(ctdb, TIMELIMIT(), options.pnn, ctdb, &reclock);
3453         if (ret != 0) {
3454                 DEBUG(DEBUG_ERR, ("Unable to get reclock file from node %u\n", options.pnn));
3455                 return ret;
3456         } else {
3457                 if (options.machinereadable){
3458                         if (reclock != NULL) {
3459                                 printf("%s", reclock);
3460                         }
3461                 } else {
3462                         if (reclock == NULL) {
3463                                 printf("No reclock file used.\n");
3464                         } else {
3465                                 printf("Reclock file:%s\n", reclock);
3466                         }
3467                 }
3468         }
3469         return 0;
3470 }
3471
3472 /*
3473   set the reclock file of a node
3474  */
3475 static int control_setreclock(struct ctdb_context *ctdb, int argc, const char **argv)
3476 {
3477         int ret;
3478         const char *reclock;
3479
3480         if (argc == 0) {
3481                 reclock = NULL;
3482         } else if (argc == 1) {
3483                 reclock = argv[0];
3484         } else {
3485                 usage();
3486         }
3487
3488         ret = ctdb_ctrl_setreclock(ctdb, TIMELIMIT(), options.pnn, reclock);
3489         if (ret != 0) {
3490                 DEBUG(DEBUG_ERR, ("Unable to get reclock file from node %u\n", options.pnn));
3491                 return ret;
3492         }
3493         return 0;
3494 }
3495
3496 /*
3497   set the natgw state on/off
3498  */
3499 static int control_setnatgwstate(struct ctdb_context *ctdb, int argc, const char **argv)
3500 {
3501         int ret;
3502         uint32_t natgwstate;
3503
3504         if (argc == 0) {
3505                 usage();
3506         }
3507
3508         if (!strcmp(argv[0], "on")) {
3509                 natgwstate = 1;
3510         } else if (!strcmp(argv[0], "off")) {
3511                 natgwstate = 0;
3512         } else {
3513                 usage();
3514         }
3515
3516         ret = ctdb_ctrl_setnatgwstate(ctdb, TIMELIMIT(), options.pnn, natgwstate);
3517         if (ret != 0) {
3518                 DEBUG(DEBUG_ERR, ("Unable to set the natgw state for node %u\n", options.pnn));
3519                 return ret;
3520         }
3521
3522         return 0;
3523 }
3524
3525 /*
3526   set the lmaster role on/off
3527  */
3528 static int control_setlmasterrole(struct ctdb_context *ctdb, int argc, const char **argv)
3529 {
3530         int ret;
3531         uint32_t lmasterrole;
3532
3533         if (argc == 0) {
3534                 usage();
3535         }
3536
3537         if (!strcmp(argv[0], "on")) {
3538                 lmasterrole = 1;
3539         } else if (!strcmp(argv[0], "off")) {
3540                 lmasterrole = 0;
3541         } else {
3542                 usage();
3543         }
3544
3545         ret = ctdb_ctrl_setlmasterrole(ctdb, TIMELIMIT(), options.pnn, lmasterrole);
3546         if (ret != 0) {
3547                 DEBUG(DEBUG_ERR, ("Unable to set the lmaster role for node %u\n", options.pnn));
3548                 return ret;
3549         }
3550
3551         return 0;
3552 }
3553
3554 /*
3555   set the recmaster role on/off
3556  */
3557 static int control_setrecmasterrole(struct ctdb_context *ctdb, int argc, const char **argv)
3558 {
3559         int ret;
3560         uint32_t recmasterrole;
3561
3562         if (argc == 0) {
3563                 usage();
3564         }
3565
3566         if (!strcmp(argv[0], "on")) {
3567                 recmasterrole = 1;
3568         } else if (!strcmp(argv[0], "off")) {
3569                 recmasterrole = 0;
3570         } else {
3571                 usage();
3572         }
3573
3574         ret = ctdb_ctrl_setrecmasterrole(ctdb, TIMELIMIT(), options.pnn, recmasterrole);
3575         if (ret != 0) {
3576                 DEBUG(DEBUG_ERR, ("Unable to set the recmaster role for node %u\n", options.pnn));
3577                 return ret;
3578         }
3579
3580         return 0;
3581 }
3582
3583 /*
3584   set debug level on a node or all nodes
3585  */
3586 static int control_setdebug(struct ctdb_context *ctdb, int argc, const char **argv)
3587 {
3588         int i, ret;
3589         int32_t level;
3590
3591         if (argc == 0) {
3592                 printf("You must specify the debug level. Valid levels are:\n");
3593                 for (i=0; debug_levels[i].description != NULL; i++) {
3594                         printf("%s (%d)\n", debug_levels[i].description, debug_levels[i].level);
3595                 }
3596
3597                 return 0;
3598         }
3599
3600         if (isalpha(argv[0][0]) || argv[0][0] == '-') {
3601                 level = get_debug_by_desc(argv[0]);
3602         } else {
3603                 level = strtol(argv[0], NULL, 0);
3604         }
3605
3606         for (i=0; debug_levels[i].description != NULL; i++) {
3607                 if (level == debug_levels[i].level) {
3608                         break;
3609                 }
3610         }
3611         if (debug_levels[i].description == NULL) {
3612                 printf("Invalid debug level, must be one of\n");
3613                 for (i=0; debug_levels[i].description != NULL; i++) {
3614                         printf("%s (%d)\n", debug_levels[i].description, debug_levels[i].level);
3615                 }
3616                 return -1;
3617         }
3618
3619         ret = ctdb_ctrl_set_debuglevel(ctdb, options.pnn, level);
3620         if (ret != 0) {
3621                 DEBUG(DEBUG_ERR, ("Unable to set debug level on node %u\n", options.pnn));
3622         }
3623         return 0;
3624 }
3625
3626
3627 /*
3628   thaw a node
3629  */
3630 static int control_thaw(struct ctdb_context *ctdb, int argc, const char **argv)
3631 {
3632         int ret;
3633         uint32_t priority;
3634
3635         if (argc == 1) {
3636                 priority = strtol(argv[0], NULL, 0);
3637         } else {
3638                 priority = 0;
3639         }
3640         DEBUG(DEBUG_ERR,("Thaw by priority %u\n", priority));
3641
3642         ret = ctdb_ctrl_thaw_priority(ctdb, TIMELIMIT(), options.pnn, priority);
3643         if (ret != 0) {
3644                 DEBUG(DEBUG_ERR, ("Unable to thaw node %u\n", options.pnn));
3645         }
3646         return 0;
3647 }
3648
3649
3650 /*
3651   attach to a database
3652  */
3653 static int control_attach(struct ctdb_context *ctdb, int argc, const char **argv)
3654 {
3655         const char *db_name;
3656         struct ctdb_db_context *ctdb_db;
3657
3658         if (argc < 1) {
3659                 usage();
3660         }
3661         db_name = argv[0];
3662
3663         ctdb_db = ctdb_attach(ctdb, db_name, false, 0);
3664         if (ctdb_db == NULL) {
3665                 DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
3666                 return -1;
3667         }
3668
3669         return 0;
3670 }
3671
3672 /*
3673   set db priority
3674  */
3675 static int control_setdbprio(struct ctdb_context *ctdb, int argc, const char **argv)
3676 {
3677         struct ctdb_db_priority db_prio;
3678         int ret;
3679
3680         if (argc < 2) {
3681                 usage();
3682         }
3683
3684         db_prio.db_id    = strtoul(argv[0], NULL, 0);
3685         db_prio.priority = strtoul(argv[1], NULL, 0);
3686
3687         ret = ctdb_ctrl_set_db_priority(ctdb, TIMELIMIT(), options.pnn, &db_prio);
3688         if (ret != 0) {
3689                 DEBUG(DEBUG_ERR,("Unable to set db prio\n"));
3690                 return -1;
3691         }
3692
3693         return 0;
3694 }
3695
3696 /*
3697   get db priority
3698  */
3699 static int control_getdbprio(struct ctdb_context *ctdb, int argc, const char **argv)
3700 {
3701         uint32_t db_id, priority;
3702         int ret;
3703
3704         if (argc < 1) {
3705                 usage();
3706         }
3707
3708         db_id = strtoul(argv[0], NULL, 0);
3709
3710         ret = ctdb_ctrl_get_db_priority(ctdb, TIMELIMIT(), options.pnn, db_id, &priority);
3711         if (ret != 0) {
3712                 DEBUG(DEBUG_ERR,("Unable to get db prio\n"));
3713                 return -1;
3714         }
3715
3716         DEBUG(DEBUG_ERR,("Priority:%u\n", priority));
3717
3718         return 0;
3719 }
3720
3721 /*
3722   run an eventscript on a node
3723  */
3724 static int control_eventscript(struct ctdb_context *ctdb, int argc, const char **argv)
3725 {
3726         TDB_DATA data;
3727         int ret;
3728         int32_t res;
3729         char *errmsg;
3730         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
3731
3732         if (argc != 1) {
3733                 DEBUG(DEBUG_ERR,("Invalid arguments\n"));
3734                 return -1;
3735         }
3736
3737         data.dptr = (unsigned char *)discard_const(argv[0]);
3738         data.dsize = strlen((char *)data.dptr) + 1;
3739
3740         DEBUG(DEBUG_ERR, ("Running eventscripts with arguments \"%s\" on node %u\n", data.dptr, options.pnn));
3741
3742         ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_RUN_EVENTSCRIPTS,
3743                            0, data, tmp_ctx, NULL, &res, NULL, &errmsg);
3744         if (ret != 0 || res != 0) {
3745                 DEBUG(DEBUG_ERR,("Failed to run eventscripts - %s\n", errmsg));
3746                 talloc_free(tmp_ctx);
3747                 return -1;
3748         }
3749         talloc_free(tmp_ctx);
3750         return 0;
3751 }
3752
3753 #define DB_VERSION 1
3754 #define MAX_DB_NAME 64
3755 struct db_file_header {
3756         unsigned long version;
3757         time_t timestamp;
3758         unsigned long persistent;
3759         unsigned long size;
3760         const char name[MAX_DB_NAME];
3761 };
3762
3763 struct backup_data {
3764         struct ctdb_marshall_buffer *records;
3765         uint32_t len;
3766         uint32_t total;
3767         bool traverse_error;
3768 };
3769
3770 static int backup_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *private)
3771 {
3772         struct backup_data *bd = talloc_get_type(private, struct backup_data);
3773         struct ctdb_rec_data *rec;
3774
3775         /* add the record */
3776         rec = ctdb_marshall_record(bd->records, 0, key, NULL, data);
3777         if (rec == NULL) {
3778                 bd->traverse_error = true;
3779                 DEBUG(DEBUG_ERR,("Failed to marshall record\n"));
3780                 return -1;
3781         }
3782         bd->records = talloc_realloc_size(NULL, bd->records, rec->length + bd->len);
3783         if (bd->records == NULL) {
3784                 DEBUG(DEBUG_ERR,("Failed to expand marshalling buffer\n"));
3785                 bd->traverse_error = true;
3786                 return -1;
3787         }
3788         bd->records->count++;
3789         memcpy(bd->len+(uint8_t *)bd->records, rec, rec->length);
3790         bd->len += rec->length;
3791         talloc_free(rec);
3792
3793         bd->total++;
3794         return 0;
3795 }
3796
3797 /*
3798  * backup a database to a file
3799  */
3800 static int control_backupdb(struct ctdb_context *ctdb, int argc, const char **argv)
3801 {
3802         int i, ret;
3803         struct ctdb_dbid_map *dbmap=NULL;
3804         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
3805         struct db_file_header dbhdr;
3806         struct ctdb_db_context *ctdb_db;
3807         struct backup_data *bd;
3808         int fh = -1;
3809         int status = -1;
3810         const char *reason = NULL;
3811
3812         if (argc != 2) {
3813                 DEBUG(DEBUG_ERR,("Invalid arguments\n"));
3814                 return -1;
3815         }
3816
3817         ret = ctdb_ctrl_getdbmap(ctdb, TIMELIMIT(), options.pnn, tmp_ctx, &dbmap);
3818         if (ret != 0) {
3819                 DEBUG(DEBUG_ERR, ("Unable to get dbids from node %u\n", options.pnn));
3820                 return ret;
3821         }
3822
3823         for(i=0;i<dbmap->num;i++){
3824                 const char *name;
3825
3826                 ctdb_ctrl_getdbname(ctdb, TIMELIMIT(), options.pnn, dbmap->dbs[i].dbid, tmp_ctx, &name);
3827                 if(!strcmp(argv[0], name)){
3828                         talloc_free(discard_const(name));
3829                         break;
3830                 }
3831                 talloc_free(discard_const(name));
3832         }
3833         if (i == dbmap->num) {
3834                 DEBUG(DEBUG_ERR,("No database with name '%s' found\n", argv[0]));
3835                 talloc_free(tmp_ctx);
3836                 return -1;
3837         }
3838
3839         ret = ctdb_ctrl_getdbhealth(ctdb, TIMELIMIT(), options.pnn,
3840                                     dbmap->dbs[i].dbid, tmp_ctx, &reason);
3841         if (ret != 0) {
3842                 DEBUG(DEBUG_ERR,("Unable to get dbhealth for database '%s'\n",
3843                                  argv[0]));
3844                 talloc_free(tmp_ctx);
3845                 return -1;
3846         }
3847         if (reason) {
3848                 uint32_t allow_unhealthy = 0;
3849
3850                 ctdb_ctrl_get_tunable(ctdb, TIMELIMIT(), options.pnn,
3851                                       "AllowUnhealthyDBRead",
3852                                       &allow_unhealthy);
3853
3854                 if (allow_unhealthy != 1) {
3855                         DEBUG(DEBUG_ERR,("database '%s' is unhealthy: %s\n",
3856                                          argv[0], reason));
3857
3858                         DEBUG(DEBUG_ERR,("disallow backup : tunnable AllowUnhealthyDBRead = %u\n",
3859                                          allow_unhealthy));
3860                         talloc_free(tmp_ctx);
3861                         return -1;
3862                 }
3863
3864                 DEBUG(DEBUG_WARNING,("WARNING database '%s' is unhealthy - see 'ctdb getdbstatus %s'\n",
3865                                      argv[0], argv[0]));
3866                 DEBUG(DEBUG_WARNING,("WARNING! allow backup of unhealthy database: "
3867                                      "tunnable AllowUnhealthyDBRead = %u\n",
3868                                      allow_unhealthy));
3869         }
3870
3871         ctdb_db = ctdb_attach(ctdb, argv[0], dbmap->dbs[i].persistent, 0);
3872         if (ctdb_db == NULL) {
3873                 DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", argv[0]));
3874                 talloc_free(tmp_ctx);
3875                 return -1;
3876         }
3877
3878
3879         ret = tdb_transaction_start(ctdb_db->ltdb->tdb);
3880         if (ret == -1) {
3881                 DEBUG(DEBUG_ERR,("Failed to start transaction\n"));
3882                 talloc_free(tmp_ctx);
3883                 return -1;
3884         }
3885
3886
3887         bd = talloc_zero(tmp_ctx, struct backup_data);
3888         if (bd == NULL) {
3889                 DEBUG(DEBUG_ERR,("Failed to allocate backup_data\n"));
3890                 talloc_free(tmp_ctx);
3891                 return -1;
3892         }
3893
3894         bd->records = talloc_zero(bd, struct ctdb_marshall_buffer);
3895         if (bd->records == NULL) {
3896                 DEBUG(DEBUG_ERR,("Failed to allocate ctdb_marshall_buffer\n"));
3897                 talloc_free(tmp_ctx);
3898                 return -1;
3899         }
3900
3901         bd->len = offsetof(struct ctdb_marshall_buffer, data);
3902         bd->records->db_id = ctdb_db->db_id;
3903         /* traverse the database collecting all records */
3904         if (tdb_traverse_read(ctdb_db->ltdb->tdb, backup_traverse, bd) == -1 ||
3905             bd->traverse_error) {
3906                 DEBUG(DEBUG_ERR,("Traverse error\n"));
3907                 talloc_free(tmp_ctx);
3908                 return -1;
3909         }
3910
3911         tdb_transaction_cancel(ctdb_db->ltdb->tdb);
3912
3913
3914         fh = open(argv[1], O_RDWR|O_CREAT, 0600);
3915         if (fh == -1) {
3916                 DEBUG(DEBUG_ERR,("Failed to open file '%s'\n", argv[1]));
3917                 talloc_free(tmp_ctx);
3918                 return -1;
3919         }
3920
3921         dbhdr.version = DB_VERSION;
3922         dbhdr.timestamp = time(NULL);
3923         dbhdr.persistent = dbmap->dbs[i].persistent;
3924         dbhdr.size = bd->len;
3925         if (strlen(argv[0]) >= MAX_DB_NAME) {
3926                 DEBUG(DEBUG_ERR,("Too long dbname\n"));
3927                 goto done;
3928         }
3929         strncpy(discard_const(dbhdr.name), argv[0], MAX_DB_NAME);
3930         ret = write(fh, &dbhdr, sizeof(dbhdr));
3931         if (ret == -1) {
3932                 DEBUG(DEBUG_ERR,("write failed: %s\n", strerror(errno)));
3933                 goto done;
3934         }
3935         ret = write(fh, bd->records, bd->len);
3936         if (ret == -1) {
3937                 DEBUG(DEBUG_ERR,("write failed: %s\n", strerror(errno)));
3938                 goto done;
3939         }
3940
3941         status = 0;
3942 done:
3943         if (fh != -1) {
3944                 ret = close(fh);
3945                 if (ret == -1) {
3946                         DEBUG(DEBUG_ERR,("close failed: %s\n", strerror(errno)));
3947                 }
3948         }
3949         talloc_free(tmp_ctx);
3950         return status;
3951 }
3952
3953 /*
3954  * restore a database from a file
3955  */
3956 static int control_restoredb(struct ctdb_context *ctdb, int argc, const char **argv)
3957 {
3958         int ret;
3959         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
3960         TDB_DATA outdata;
3961         TDB_DATA data;
3962         struct db_file_header dbhdr;
3963         struct ctdb_db_context *ctdb_db;
3964         struct ctdb_node_map *nodemap=NULL;
3965         struct ctdb_vnn_map *vnnmap=NULL;
3966         int i, fh;
3967         struct ctdb_control_wipe_database w;
3968         uint32_t *nodes;
3969         uint32_t generation;
3970         struct tm *tm;
3971         char tbuf[100];
3972         char *dbname;
3973
3974         if (argc < 1 || argc > 2) {
3975                 DEBUG(DEBUG_ERR,("Invalid arguments\n"));
3976                 return -1;
3977         }
3978
3979         fh = open(argv[0], O_RDONLY);
3980         if (fh == -1) {
3981                 DEBUG(DEBUG_ERR,("Failed to open file '%s'\n", argv[0]));
3982                 talloc_free(tmp_ctx);
3983                 return -1;
3984         }
3985
3986         read(fh, &dbhdr, sizeof(dbhdr));
3987         if (dbhdr.version != DB_VERSION) {
3988                 DEBUG(DEBUG_ERR,("Invalid version of database dump. File is version %lu but expected version was %u\n", dbhdr.version, DB_VERSION));
3989                 talloc_free(tmp_ctx);
3990                 return -1;
3991         }
3992
3993         dbname = discard_const(dbhdr.name);
3994         if (argc == 2) {
3995                 dbname = discard_const(argv[1]);
3996         }
3997
3998         outdata.dsize = dbhdr.size;
3999         outdata.dptr = talloc_size(tmp_ctx, outdata.dsize);
4000         if (outdata.dptr == NULL) {
4001                 DEBUG(DEBUG_ERR,("Failed to allocate data of size '%lu'\n", dbhdr.size));
4002                 close(fh);
4003                 talloc_free(tmp_ctx);
4004                 return -1;
4005         }
4006         read(fh, outdata.dptr, outdata.dsize);
4007         close(fh);
4008
4009         tm = localtime(&dbhdr.timestamp);
4010         strftime(tbuf,sizeof(tbuf)-1,"%Y/%m/%d %H:%M:%S", tm);
4011         printf("Restoring database '%s' from backup @ %s\n",
4012                 dbname, tbuf);
4013
4014
4015         ctdb_db = ctdb_attach(ctdb, dbname, dbhdr.persistent, 0);
4016         if (ctdb_db == NULL) {
4017                 DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", dbname));
4018                 talloc_free(tmp_ctx);
4019                 return -1;
4020         }
4021
4022         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap);
4023         if (ret != 0) {
4024                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
4025                 talloc_free(tmp_ctx);
4026                 return ret;
4027         }
4028
4029
4030         ret = ctdb_ctrl_getvnnmap(ctdb, TIMELIMIT(), options.pnn, tmp_ctx, &vnnmap);
4031         if (ret != 0) {
4032                 DEBUG(DEBUG_ERR, ("Unable to get vnnmap from node %u\n", options.pnn));
4033                 talloc_free(tmp_ctx);
4034                 return ret;
4035         }
4036
4037         /* freeze all nodes */
4038         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4039         for (i=1; i<=NUM_DB_PRIORITIES; i++) {
4040                 if (ctdb_client_async_control(ctdb, CTDB_CONTROL_FREEZE,
4041                                         nodes, i,
4042                                         TIMELIMIT(),
4043                                         false, tdb_null,
4044                                         NULL, NULL,
4045                                         NULL) != 0) {
4046                         DEBUG(DEBUG_ERR, ("Unable to freeze nodes.\n"));
4047                         ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4048                         talloc_free(tmp_ctx);
4049                         return -1;
4050                 }
4051         }
4052
4053         generation = vnnmap->generation;
4054         data.dptr = (void *)&generation;
4055         data.dsize = sizeof(generation);
4056
4057         /* start a cluster wide transaction */
4058         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4059         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_TRANSACTION_START,
4060                                         nodes, 0,
4061                                         TIMELIMIT(), false, data,
4062                                         NULL, NULL,
4063                                         NULL) != 0) {
4064                 DEBUG(DEBUG_ERR, ("Unable to start cluster wide transactions.\n"));
4065                 return -1;
4066         }
4067
4068
4069         w.db_id = ctdb_db->db_id;
4070         w.transaction_id = generation;
4071
4072         data.dptr = (void *)&w;
4073         data.dsize = sizeof(w);
4074
4075         /* wipe all the remote databases. */
4076         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4077         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_WIPE_DATABASE,
4078                                         nodes, 0,
4079                                         TIMELIMIT(), false, data,
4080                                         NULL, NULL,
4081                                         NULL) != 0) {
4082                 DEBUG(DEBUG_ERR, ("Unable to wipe database.\n"));
4083                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4084                 talloc_free(tmp_ctx);
4085                 return -1;
4086         }
4087
4088         /* push the database */
4089         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4090         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_PUSH_DB,
4091                                         nodes, 0,
4092                                         TIMELIMIT(), false, outdata,
4093                                         NULL, NULL,
4094                                         NULL) != 0) {
4095                 DEBUG(DEBUG_ERR, ("Failed to push database.\n"));
4096                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4097                 talloc_free(tmp_ctx);
4098                 return -1;
4099         }
4100
4101         data.dptr = (void *)&ctdb_db->db_id;
4102         data.dsize = sizeof(ctdb_db->db_id);
4103
4104         /* mark the database as healthy */
4105         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4106         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_DB_SET_HEALTHY,
4107                                         nodes, 0,
4108                                         TIMELIMIT(), false, data,
4109                                         NULL, NULL,
4110                                         NULL) != 0) {
4111                 DEBUG(DEBUG_ERR, ("Failed to mark database as healthy.\n"));
4112                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4113                 talloc_free(tmp_ctx);
4114                 return -1;
4115         }
4116
4117         data.dptr = (void *)&generation;
4118         data.dsize = sizeof(generation);
4119
4120         /* commit all the changes */
4121         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_TRANSACTION_COMMIT,
4122                                         nodes, 0,
4123                                         TIMELIMIT(), false, data,
4124                                         NULL, NULL,
4125                                         NULL) != 0) {
4126                 DEBUG(DEBUG_ERR, ("Unable to commit databases.\n"));
4127                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4128                 talloc_free(tmp_ctx);
4129                 return -1;
4130         }
4131
4132
4133         /* thaw all nodes */
4134         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4135         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_THAW,
4136                                         nodes, 0,
4137                                         TIMELIMIT(),
4138                                         false, tdb_null,
4139                                         NULL, NULL,
4140                                         NULL) != 0) {
4141                 DEBUG(DEBUG_ERR, ("Unable to thaw nodes.\n"));
4142                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4143                 talloc_free(tmp_ctx);
4144                 return -1;
4145         }
4146
4147
4148         talloc_free(tmp_ctx);
4149         return 0;
4150 }
4151
4152 /*
4153  * dump a database backup from a file
4154  */
4155 static int control_dumpdbbackup(struct ctdb_context *ctdb, int argc, const char **argv)
4156 {
4157         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
4158         TDB_DATA outdata;
4159         struct db_file_header dbhdr;
4160         int i, fh;
4161         struct tm *tm;
4162         char tbuf[100];
4163         struct ctdb_rec_data *rec = NULL;
4164         struct ctdb_marshall_buffer *m;
4165
4166         if (argc != 1) {
4167                 DEBUG(DEBUG_ERR,("Invalid arguments\n"));
4168                 return -1;
4169         }
4170
4171         fh = open(argv[0], O_RDONLY);
4172         if (fh == -1) {
4173                 DEBUG(DEBUG_ERR,("Failed to open file '%s'\n", argv[0]));
4174                 talloc_free(tmp_ctx);
4175                 return -1;
4176         }
4177
4178         read(fh, &dbhdr, sizeof(dbhdr));
4179         if (dbhdr.version != DB_VERSION) {
4180                 DEBUG(DEBUG_ERR,("Invalid version of database dump. File is version %lu but expected version was %u\n", dbhdr.version, DB_VERSION));
4181                 talloc_free(tmp_ctx);
4182                 return -1;
4183         }
4184
4185         outdata.dsize = dbhdr.size;
4186         outdata.dptr = talloc_size(tmp_ctx, outdata.dsize);
4187         if (outdata.dptr == NULL) {
4188                 DEBUG(DEBUG_ERR,("Failed to allocate data of size '%lu'\n", dbhdr.size));
4189                 close(fh);
4190                 talloc_free(tmp_ctx);
4191                 return -1;
4192         }
4193         read(fh, outdata.dptr, outdata.dsize);
4194         close(fh);
4195         m = (struct ctdb_marshall_buffer *)outdata.dptr;
4196
4197         tm = localtime(&dbhdr.timestamp);
4198         strftime(tbuf,sizeof(tbuf)-1,"%Y/%m/%d %H:%M:%S", tm);
4199         printf("Backup of database name:'%s' dbid:0x%x08x from @ %s\n",
4200                 dbhdr.name, m->db_id, tbuf);
4201
4202         for (i=0; i < m->count; i++) {
4203                 uint32_t reqid = 0;
4204                 TDB_DATA key, data;
4205
4206                 /* we do not want the header splitted, so we pass NULL*/
4207                 rec = ctdb_marshall_loop_next(m, rec, &reqid,
4208                                               NULL, &key, &data);
4209
4210                 ctdb_dumpdb_record(ctdb, key, data, stdout);
4211         }
4212
4213         printf("Dumped %d records\n", i);
4214         talloc_free(tmp_ctx);
4215         return 0;
4216 }
4217
4218 /*
4219  * wipe a database from a file
4220  */
4221 static int control_wipedb(struct ctdb_context *ctdb, int argc,
4222                           const char **argv)
4223 {
4224         int ret;
4225         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
4226         TDB_DATA data;
4227         struct ctdb_db_context *ctdb_db;
4228         struct ctdb_node_map *nodemap = NULL;
4229         struct ctdb_vnn_map *vnnmap = NULL;
4230         int i;
4231         struct ctdb_control_wipe_database w;
4232         uint32_t *nodes;
4233         uint32_t generation;
4234         struct ctdb_dbid_map *dbmap = NULL;
4235
4236         if (argc != 1) {
4237                 DEBUG(DEBUG_ERR,("Invalid arguments\n"));
4238                 return -1;
4239         }
4240
4241         ret = ctdb_ctrl_getdbmap(ctdb, TIMELIMIT(), options.pnn, tmp_ctx,
4242                                  &dbmap);
4243         if (ret != 0) {
4244                 DEBUG(DEBUG_ERR, ("Unable to get dbids from node %u\n",
4245                                   options.pnn));
4246                 return ret;
4247         }
4248
4249         for(i=0;i<dbmap->num;i++){
4250                 const char *name;
4251
4252                 ctdb_ctrl_getdbname(ctdb, TIMELIMIT(), options.pnn,
4253                                     dbmap->dbs[i].dbid, tmp_ctx, &name);
4254                 if(!strcmp(argv[0], name)){
4255                         talloc_free(discard_const(name));
4256                         break;
4257                 }
4258                 talloc_free(discard_const(name));
4259         }
4260         if (i == dbmap->num) {
4261                 DEBUG(DEBUG_ERR, ("No database with name '%s' found\n",
4262                                   argv[0]));
4263                 talloc_free(tmp_ctx);
4264                 return -1;
4265         }
4266
4267         ctdb_db = ctdb_attach(ctdb, argv[0], dbmap->dbs[i].persistent, 0);
4268         if (ctdb_db == NULL) {
4269                 DEBUG(DEBUG_ERR, ("Unable to attach to database '%s'\n",
4270                                   argv[0]));
4271                 talloc_free(tmp_ctx);
4272                 return -1;
4273         }
4274
4275         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb,
4276                                    &nodemap);
4277         if (ret != 0) {
4278                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n",
4279                                   options.pnn));
4280                 talloc_free(tmp_ctx);
4281                 return ret;
4282         }
4283
4284         ret = ctdb_ctrl_getvnnmap(ctdb, TIMELIMIT(), options.pnn, tmp_ctx,
4285                                   &vnnmap);
4286         if (ret != 0) {
4287                 DEBUG(DEBUG_ERR, ("Unable to get vnnmap from node %u\n",
4288                                   options.pnn));
4289                 talloc_free(tmp_ctx);
4290                 return ret;
4291         }
4292
4293         /* freeze all nodes */
4294         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4295         for (i=1; i<=NUM_DB_PRIORITIES; i++) {
4296                 ret = ctdb_client_async_control(ctdb, CTDB_CONTROL_FREEZE,
4297                                                 nodes, i,
4298                                                 TIMELIMIT(),
4299                                                 false, tdb_null,
4300                                                 NULL, NULL,
4301                                                 NULL);
4302                 if (ret != 0) {
4303                         DEBUG(DEBUG_ERR, ("Unable to freeze nodes.\n"));
4304                         ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn,
4305                                              CTDB_RECOVERY_ACTIVE);
4306                         talloc_free(tmp_ctx);
4307                         return -1;
4308                 }
4309         }
4310
4311         generation = vnnmap->generation;
4312         data.dptr = (void *)&generation;
4313         data.dsize = sizeof(generation);
4314
4315         /* start a cluster wide transaction */
4316         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4317         ret = ctdb_client_async_control(ctdb, CTDB_CONTROL_TRANSACTION_START,
4318                                         nodes, 0,
4319                                         TIMELIMIT(), false, data,
4320                                         NULL, NULL,
4321                                         NULL);
4322         if (ret!= 0) {
4323                 DEBUG(DEBUG_ERR, ("Unable to start cluster wide "
4324                                   "transactions.\n"));
4325                 return -1;
4326         }
4327
4328         w.db_id = ctdb_db->db_id;
4329         w.transaction_id = generation;
4330
4331         data.dptr = (void *)&w;
4332         data.dsize = sizeof(w);
4333
4334         /* wipe all the remote databases. */
4335         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4336         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_WIPE_DATABASE,
4337                                         nodes, 0,
4338                                         TIMELIMIT(), false, data,
4339                                         NULL, NULL,
4340                                         NULL) != 0) {
4341                 DEBUG(DEBUG_ERR, ("Unable to wipe database.\n"));
4342                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4343                 talloc_free(tmp_ctx);
4344                 return -1;
4345         }
4346
4347         data.dptr = (void *)&ctdb_db->db_id;
4348         data.dsize = sizeof(ctdb_db->db_id);
4349
4350         /* mark the database as healthy */
4351         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4352         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_DB_SET_HEALTHY,
4353                                         nodes, 0,
4354                                         TIMELIMIT(), false, data,
4355                                         NULL, NULL,
4356                                         NULL) != 0) {
4357                 DEBUG(DEBUG_ERR, ("Failed to mark database as healthy.\n"));
4358                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4359                 talloc_free(tmp_ctx);
4360                 return -1;
4361         }
4362
4363         data.dptr = (void *)&generation;
4364         data.dsize = sizeof(generation);
4365
4366         /* commit all the changes */
4367         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_TRANSACTION_COMMIT,
4368                                         nodes, 0,
4369                                         TIMELIMIT(), false, data,
4370                                         NULL, NULL,
4371                                         NULL) != 0) {
4372                 DEBUG(DEBUG_ERR, ("Unable to commit databases.\n"));
4373                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4374                 talloc_free(tmp_ctx);
4375                 return -1;
4376         }
4377
4378         /* thaw all nodes */
4379         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
4380         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_THAW,
4381                                         nodes, 0,
4382                                         TIMELIMIT(),
4383                                         false, tdb_null,
4384                                         NULL, NULL,
4385                                         NULL) != 0) {
4386                 DEBUG(DEBUG_ERR, ("Unable to thaw nodes.\n"));
4387                 ctdb_ctrl_setrecmode(ctdb, TIMELIMIT(), options.pnn, CTDB_RECOVERY_ACTIVE);
4388                 talloc_free(tmp_ctx);
4389                 return -1;
4390         }
4391
4392         talloc_free(tmp_ctx);
4393         return 0;
4394 }
4395
4396 /*
4397  * set flags of a node in the nodemap
4398  */
4399 static int control_setflags(struct ctdb_context *ctdb, int argc, const char **argv)
4400 {
4401         int ret;
4402         int32_t status;
4403         int node;
4404         int flags;
4405         TDB_DATA data;
4406         struct ctdb_node_flag_change c;
4407
4408         if (argc != 2) {
4409                 usage();
4410                 return -1;
4411         }
4412
4413         if (sscanf(argv[0], "%d", &node) != 1) {
4414                 DEBUG(DEBUG_ERR, ("Badly formed node\n"));
4415                 usage();
4416                 return -1;
4417         }
4418         if (sscanf(argv[1], "0x%x", &flags) != 1) {
4419                 DEBUG(DEBUG_ERR, ("Badly formed flags\n"));
4420                 usage();
4421                 return -1;
4422         }
4423
4424         c.pnn       = node;
4425         c.old_flags = 0;
4426         c.new_flags = flags;
4427
4428         data.dsize = sizeof(c);
4429         data.dptr = (unsigned char *)&c;
4430
4431         ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_MODIFY_FLAGS, 0,
4432                            data, NULL, NULL, &status, NULL, NULL);
4433         if (ret != 0 || status != 0) {
4434                 DEBUG(DEBUG_ERR,("Failed to modify flags\n"));
4435                 return -1;
4436         }
4437         return 0;
4438 }
4439
4440 /*
4441   dump memory usage
4442  */
4443 static int control_dumpmemory(struct ctdb_context *ctdb, int argc, const char **argv)
4444 {
4445         TDB_DATA data;
4446         int ret;
4447         int32_t res;
4448         char *errmsg;
4449         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
4450         ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_DUMP_MEMORY,
4451                            0, tdb_null, tmp_ctx, &data, &res, NULL, &errmsg);
4452         if (ret != 0 || res != 0) {
4453                 DEBUG(DEBUG_ERR,("Failed to dump memory - %s\n", errmsg));
4454                 talloc_free(tmp_ctx);
4455                 return -1;
4456         }
4457         write(1, data.dptr, data.dsize);
4458         talloc_free(tmp_ctx);
4459         return 0;
4460 }
4461
4462 /*
4463   handler for memory dumps
4464 */
4465 static void mem_dump_handler(struct ctdb_context *ctdb, uint64_t srvid,
4466                              TDB_DATA data, void *private_data)
4467 {
4468         write(1, data.dptr, data.dsize);
4469         exit(0);
4470 }
4471
4472 /*
4473   dump memory usage on the recovery daemon
4474  */
4475 static int control_rddumpmemory(struct ctdb_context *ctdb, int argc, const char **argv)
4476 {
4477         int ret;
4478         TDB_DATA data;
4479         struct rd_memdump_reply rd;
4480
4481         rd.pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
4482         if (rd.pnn == -1) {
4483                 DEBUG(DEBUG_ERR, ("Failed to get pnn of local node\n"));
4484                 return -1;
4485         }
4486         rd.srvid = getpid();
4487
4488         /* register a message port for receiveing the reply so that we
4489            can receive the reply
4490         */
4491         ctdb_client_set_message_handler(ctdb, rd.srvid, mem_dump_handler, NULL);
4492
4493
4494         data.dptr = (uint8_t *)&rd;
4495         data.dsize = sizeof(rd);
4496
4497         ret = ctdb_client_send_message(ctdb, options.pnn, CTDB_SRVID_MEM_DUMP, data);
4498         if (ret != 0) {
4499                 DEBUG(DEBUG_ERR,("Failed to send memdump request message to %u\n", options.pnn));
4500                 return -1;
4501         }
4502
4503         /* this loop will terminate when we have received the reply */
4504         while (1) {
4505                 event_loop_once(ctdb->ev);
4506         }
4507
4508         return 0;
4509 }
4510
4511 /*
4512   send a message to a srvid
4513  */
4514 static int control_msgsend(struct ctdb_context *ctdb, int argc, const char **argv)
4515 {
4516         unsigned long srvid;
4517         int ret;
4518         TDB_DATA data;
4519
4520         if (argc < 2) {
4521                 usage();
4522         }
4523
4524         srvid      = strtoul(argv[0], NULL, 0);
4525
4526         data.dptr = (uint8_t *)discard_const(argv[1]);
4527         data.dsize= strlen(argv[1]);
4528
4529         ret = ctdb_client_send_message(ctdb, CTDB_BROADCAST_CONNECTED, srvid, data);
4530         if (ret != 0) {
4531                 DEBUG(DEBUG_ERR,("Failed to send memdump request message to %u\n", options.pnn));
4532                 return -1;
4533         }
4534
4535         return 0;
4536 }
4537
4538 /*
4539   handler for msglisten
4540 */
4541 static void msglisten_handler(struct ctdb_context *ctdb, uint64_t srvid,
4542                              TDB_DATA data, void *private_data)
4543 {
4544         int i;
4545
4546         printf("Message received: ");
4547         for (i=0;i<data.dsize;i++) {
4548                 printf("%c", data.dptr[i]);
4549         }
4550         printf("\n");
4551 }
4552
4553 /*
4554   listen for messages on a messageport
4555  */
4556 static int control_msglisten(struct ctdb_context *ctdb, int argc, const char **argv)
4557 {
4558         uint64_t srvid;
4559
4560         srvid = getpid();
4561
4562         /* register a message port and listen for messages
4563         */
4564         ctdb_client_set_message_handler(ctdb, srvid, msglisten_handler, NULL);
4565         printf("Listening for messages on srvid:%d\n", (int)srvid);
4566
4567         while (1) {
4568                 event_loop_once(ctdb->ev);
4569         }
4570
4571         return 0;
4572 }
4573
4574 /*
4575   list all nodes in the cluster
4576   we parse the nodes file directly
4577  */
4578 static int control_listnodes(struct ctdb_context *ctdb, int argc, const char **argv)
4579 {
4580         TALLOC_CTX *mem_ctx = talloc_new(NULL);
4581         struct pnn_node *pnn_nodes;
4582         struct pnn_node *pnn_node;
4583
4584         pnn_nodes = read_nodes_file(mem_ctx);
4585         if (pnn_nodes == NULL) {
4586                 DEBUG(DEBUG_ERR,("Failed to read nodes file\n"));
4587                 talloc_free(mem_ctx);
4588                 return -1;
4589         }
4590
4591         for(pnn_node=pnn_nodes;pnn_node;pnn_node=pnn_node->next) {
4592                 ctdb_sock_addr addr;
4593                 if (parse_ip(pnn_node->addr, NULL, 63999, &addr) == 0) {
4594                         DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s' in nodes file\n", pnn_node->addr));
4595                         talloc_free(mem_ctx);
4596                         return -1;
4597                 }
4598                 if (options.machinereadable){
4599                         printf(":%d:%s:\n", pnn_node->pnn, pnn_node->addr);
4600                 } else {
4601                         printf("%s\n", pnn_node->addr);
4602                 }
4603         }
4604         talloc_free(mem_ctx);
4605
4606         return 0;
4607 }
4608
4609 /*
4610   reload the nodes file on the local node
4611  */
4612 static int control_reload_nodes_file(struct ctdb_context *ctdb, int argc, const char **argv)
4613 {
4614         int i, ret;
4615         int mypnn;
4616         struct ctdb_node_map *nodemap=NULL;
4617
4618         mypnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
4619         if (mypnn == -1) {
4620                 DEBUG(DEBUG_ERR, ("Failed to read pnn of local node\n"));
4621                 return -1;
4622         }
4623
4624         ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap);
4625         if (ret != 0) {
4626                 DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
4627                 return ret;
4628         }
4629
4630         /* reload the nodes file on all remote nodes */
4631         for (i=0;i<nodemap->num;i++) {
4632                 if (nodemap->nodes[i].pnn == mypnn) {
4633                         continue;
4634                 }
4635                 DEBUG(DEBUG_NOTICE, ("Reloading nodes file on node %u\n", nodemap->nodes[i].pnn));
4636                 ret = ctdb_ctrl_reload_nodes_file(ctdb, TIMELIMIT(),
4637                         nodemap->nodes[i].pnn);
4638                 if (ret != 0) {
4639                         DEBUG(DEBUG_ERR, ("ERROR: Failed to reload nodes file on node %u. You MUST fix that node manually!\n", nodemap->nodes[i].pnn));
4640                 }
4641         }
4642
4643         /* reload the nodes file on the local node */
4644         DEBUG(DEBUG_NOTICE, ("Reloading nodes file on node %u\n", mypnn));
4645         ret = ctdb_ctrl_reload_nodes_file(ctdb, TIMELIMIT(), mypnn);
4646         if (ret != 0) {
4647                 DEBUG(DEBUG_ERR, ("ERROR: Failed to reload nodes file on node %u. You MUST fix that node manually!\n", mypnn));
4648         }
4649
4650         /* initiate a recovery */
4651         control_recover(ctdb, argc, argv);
4652
4653         return 0;
4654 }
4655
4656
4657 static const struct {
4658         const char *name;
4659         int (*fn)(struct ctdb_context *, int, const char **);
4660         bool auto_all;
4661         bool without_daemon; /* can be run without daemon running ? */
4662         const char *msg;
4663         const char *args;
4664 } ctdb_commands[] = {
4665 #ifdef CTDB_VERS
4666         { "version",         control_version,           true,   false,  "show version of ctdb" },
4667 #endif
4668         { "status",          control_status,            true,   false,  "show node status" },
4669         { "uptime",          control_uptime,            true,   false,  "show node uptime" },
4670         { "ping",            control_ping,              true,   false,  "ping all nodes" },
4671         { "getvar",          control_getvar,            true,   false,  "get a tunable variable",               "<name>"},
4672         { "setvar",          control_setvar,            true,   false,  "set a tunable variable",               "<name> <value>"},
4673         { "listvars",        control_listvars,          true,   false,  "list tunable variables"},
4674         { "statistics",      control_statistics,        false,  false, "show statistics" },
4675         { "statisticsreset", control_statistics_reset,  true,   false,  "reset statistics"},
4676         { "stats",           control_stats,             false,  false,  "show rolling statistics", "[number of history records]" },
4677         { "ip",              control_ip,                false,  false,  "show which public ip's that ctdb manages" },
4678         { "ipinfo",          control_ipinfo,            true,   false,  "show details about a public ip that ctdb manages", "<ip>" },
4679         { "ifaces",          control_ifaces,            true,   false,  "show which interfaces that ctdb manages" },
4680         { "setifacelink",    control_setifacelink,      true,   false,  "set interface link status", "<iface> <status>" },
4681         { "process-exists",  control_process_exists,    true,   false,  "check if a process exists on a node",  "<pid>"},
4682         { "getdbmap",        control_getdbmap,          true,   false,  "show the database map" },
4683         { "getdbstatus",     control_getdbstatus,       true,   false,  "show the status of a database", "<dbname>" },
4684         { "catdb",           control_catdb,             true,   false,  "dump a database" ,                     "<dbname>"},
4685         { "getmonmode",      control_getmonmode,        true,   false,  "show monitoring mode" },
4686         { "getcapabilities", control_getcapabilities,   true,   false,  "show node capabilities" },
4687         { "pnn",             control_pnn,               true,   false,  "show the pnn of the currnet node" },
4688         { "lvs",             control_lvs,               true,   false,  "show lvs configuration" },
4689         { "lvsmaster",       control_lvsmaster,         true,   false,  "show which node is the lvs master" },
4690         { "disablemonitor",      control_disable_monmode,true,  false,  "set monitoring mode to DISABLE" },
4691         { "enablemonitor",      control_enable_monmode, true,   false,  "set monitoring mode to ACTIVE" },
4692         { "setdebug",        control_setdebug,          true,   false,  "set debug level",                      "<EMERG|ALERT|CRIT|ERR|WARNING|NOTICE|INFO|DEBUG>" },
4693         { "getdebug",        control_getdebug,          true,   false,  "get debug level" },
4694         { "getlog",          control_getlog,            true,   false,  "get the log data from the in memory ringbuffer", "<level>" },
4695         { "clearlog",          control_clearlog,        true,   false,  "clear the log data from the in memory ringbuffer" },
4696         { "attach",          control_attach,            true,   false,  "attach to a database",                 "<dbname>" },
4697         { "dumpmemory",      control_dumpmemory,        true,   false,  "dump memory map to stdout" },
4698         { "rddumpmemory",    control_rddumpmemory,      true,   false,  "dump memory map from the recovery daemon to stdout" },
4699         { "getpid",          control_getpid,            true,   false,  "get ctdbd process ID" },
4700         { "disable",         control_disable,           true,   false,  "disable a nodes public IP" },
4701         { "enable",          control_enable,            true,   false,  "enable a nodes public IP" },
4702         { "stop",            control_stop,              true,   false,  "stop a node" },
4703         { "continue",        control_continue,          true,   false,  "re-start a stopped node" },
4704         { "ban",             control_ban,               true,   false,  "ban a node from the cluster",          "<bantime|0>"},
4705         { "unban",           control_unban,             true,   false,  "unban a node" },
4706         { "showban",         control_showban,           true,   false,  "show ban information"},
4707         { "shutdown",        control_shutdown,          true,   false,  "shutdown ctdbd" },
4708         { "recover",         control_recover,           true,   false,  "force recovery" },
4709         { "sync",            control_ipreallocate,      true,   false,  "wait until ctdbd has synced all state changes" },
4710         { "ipreallocate",    control_ipreallocate,      true,   false,  "force the recovery daemon to perform a ip reallocation procedure" },
4711         { "thaw",            control_thaw,              true,   false,  "thaw databases", "[priority:1-3]" },
4712         { "isnotrecmaster",  control_isnotrecmaster,    false,  false,  "check if the local node is recmaster or not" },
4713         { "killtcp",         kill_tcp,                  false,  false, "kill a tcp connection.", "<srcip:port> <dstip:port>" },
4714         { "gratiousarp",     control_gratious_arp,      false,  false, "send a gratious arp", "<ip> <interface>" },
4715         { "tickle",          tickle_tcp,                false,  false, "send a tcp tickle ack", "<srcip:port> <dstip:port>" },
4716         { "gettickles",      control_get_tickles,       false,  false, "get the list of tickles registered for this ip", "<ip> [<port>]" },
4717         { "addtickle",       control_add_tickle,        false,  false, "add a tickle for this ip", "<ip>:<port> <ip>:<port>" },
4718
4719         { "deltickle",       control_del_tickle,        false,  false, "delete a tickle from this ip", "<ip>:<port> <ip>:<port>" },
4720
4721         { "regsrvid",        regsrvid,                  false,  false, "register a server id", "<pnn> <type> <id>" },
4722         { "unregsrvid",      unregsrvid,                false,  false, "unregister a server id", "<pnn> <type> <id>" },
4723         { "chksrvid",        chksrvid,                  false,  false, "check if a server id exists", "<pnn> <type> <id>" },
4724         { "getsrvids",       getsrvids,                 false,  false, "get a list of all server ids"},
4725         { "vacuum",          ctdb_vacuum,               false,  false, "vacuum the databases of empty records", "[max_records]"},
4726         { "repack",          ctdb_repack,               false,  false, "repack all databases", "[max_freelist]"},
4727         { "listnodes",       control_listnodes,         false,  true, "list all nodes in the cluster"},
4728         { "reloadnodes",     control_reload_nodes_file, false,  false, "reload the nodes file and restart the transport on all nodes"},
4729         { "moveip",          control_moveip,            false,  false, "move/failover an ip address to another node", "<ip> <node>"},
4730         { "addip",           control_addip,             true,   false, "add a ip address to a node", "<ip/mask> <iface>"},
4731         { "delip",           control_delip,             false,  false, "delete an ip address from a node", "<ip>"},
4732         { "eventscript",     control_eventscript,       true,   false, "run the eventscript with the given parameters on a node", "<arguments>"},
4733         { "backupdb",        control_backupdb,          false,  false, "backup the database into a file.", "<database> <file>"},
4734         { "restoredb",        control_restoredb,        false,  false, "restore the database from a file.", "<file> [dbname]"},
4735         { "dumpdbbackup",    control_dumpdbbackup,      false,  true,  "dump database backup from a file.", "<file>"},
4736         { "wipedb",           control_wipedb,        false,     false, "wipe the contents of a database.", "<dbname>"},
4737         { "recmaster",        control_recmaster,        false,  false, "show the pnn for the recovery master."},
4738         { "setflags",        control_setflags,          false,  false, "set flags for a node in the nodemap.", "<node> <flags>"},
4739         { "scriptstatus",    control_scriptstatus,  false,      false, "show the status of the monitoring scripts (or all scripts)", "[all]"},
4740         { "enablescript",     control_enablescript,  false,     false, "enable an eventscript", "<script>"},
4741         { "disablescript",    control_disablescript,  false,    false, "disable an eventscript", "<script>"},
4742         { "natgwlist",        control_natgwlist,        false,  false, "show the nodes belonging to this natgw configuration"},
4743         { "xpnn",             control_xpnn,             true,   true,  "find the pnn of the local node without talking to the daemon (unreliable)" },
4744         { "getreclock",       control_getreclock,       false,  false, "Show the reclock file of a node"},
4745         { "setreclock",       control_setreclock,       false,  false, "Set/clear the reclock file of a node", "[filename]"},
4746         { "setnatgwstate",    control_setnatgwstate,    false,  false, "Set NATGW state to on/off", "{on|off}"},
4747         { "setlmasterrole",   control_setlmasterrole,   false,  false, "Set LMASTER role to on/off", "{on|off}"},
4748         { "setrecmasterrole", control_setrecmasterrole, false,  false, "Set RECMASTER role to on/off", "{on|off}"},
4749         { "setdbprio",        control_setdbprio,        false,  false, "Set DB priority", "<dbid> <prio:1-3>"},
4750         { "getdbprio",        control_getdbprio,        false,  false, "Get DB priority", "<dbid>"},
4751         { "msglisten",        control_msglisten,        false,  false, "Listen on a srvid port for messages", "<msg srvid>"},
4752         { "msgsend",          control_msgsend,  false,  false, "Send a message to srvid", "<srvid> <message>"},
4753         { "sync",            control_ipreallocate,      false,  false,  "wait until ctdbd has synced all state changes" },
4754         { "pfetch",          control_pfetch,            false,  false,  "fetch a record from a persistent database", "<db> <key> [<file>]" },
4755         { "pstore",          control_pstore,            false,  false,  "write a record to a persistent database", "<db> <key> <file containing record>" },
4756         { "tfetch",          control_tfetch,            false,  true,  "fetch a record from a [c]tdb-file", "<tdb-file> <key> [<file>]" },
4757 };
4758
4759 /*
4760   show usage message
4761  */
4762 static void usage(void)
4763 {
4764         int i;
4765         printf(
4766 "Usage: ctdb [options] <control>\n" \
4767 "Options:\n" \
4768 "   -n <node>          choose node number, or 'all' (defaults to local node)\n"
4769 "   -Y                 generate machinereadable output\n"
4770 "   -v                 generate verbose output\n"
4771 "   -t <timelimit>     set timelimit for control in seconds (default %u)\n", options.timelimit);
4772         printf("Controls:\n");
4773         for (i=0;i<ARRAY_SIZE(ctdb_commands);i++) {
4774                 printf("  %-15s %-27s  %s\n",
4775                        ctdb_commands[i].name,
4776                        ctdb_commands[i].args?ctdb_commands[i].args:"",
4777                        ctdb_commands[i].msg);
4778         }
4779         exit(1);
4780 }
4781
4782
4783 static void ctdb_alarm(int sig)
4784 {
4785         printf("Maximum runtime exceeded - exiting\n");
4786         _exit(ERR_TIMEOUT);
4787 }
4788
4789 /*
4790   main program
4791 */
4792 int main(int argc, const char *argv[])
4793 {
4794         struct ctdb_context *ctdb;
4795         char *nodestring = NULL;
4796         struct poptOption popt_options[] = {
4797                 POPT_AUTOHELP
4798                 POPT_CTDB_CMDLINE
4799                 { "timelimit", 't', POPT_ARG_INT, &options.timelimit, 0, "timelimit", "integer" },
4800                 { "node",      'n', POPT_ARG_STRING, &nodestring, 0, "node", "integer|all" },
4801                 { "machinereadable", 'Y', POPT_ARG_NONE, &options.machinereadable, 0, "enable machinereadable output", NULL },
4802                 { "verbose",    'v', POPT_ARG_NONE, &options.verbose, 0, "enable verbose output", NULL },
4803                 { "maxruntime", 'T', POPT_ARG_INT, &options.maxruntime, 0, "die if runtime exceeds this limit (in seconds)", "integer" },
4804                 POPT_TABLEEND
4805         };
4806         int opt;
4807         const char **extra_argv;
4808         int extra_argc = 0;
4809         int ret=-1, i;
4810         poptContext pc;
4811         struct event_context *ev;
4812         const char *control;
4813
4814         setlinebuf(stdout);
4815
4816         /* set some defaults */
4817         options.maxruntime = 0;
4818         options.timelimit = 3;
4819         options.pnn = CTDB_CURRENT_NODE;
4820
4821         pc = poptGetContext(argv[0], argc, argv, popt_options, POPT_CONTEXT_KEEP_FIRST);
4822
4823         while ((opt = poptGetNextOpt(pc)) != -1) {
4824                 switch (opt) {
4825                 default:
4826                         DEBUG(DEBUG_ERR, ("Invalid option %s: %s\n",
4827                                 poptBadOption(pc, 0), poptStrerror(opt)));
4828                         exit(1);
4829                 }
4830         }
4831
4832         /* setup the remaining options for the main program to use */
4833         extra_argv = poptGetArgs(pc);
4834         if (extra_argv) {
4835                 extra_argv++;
4836                 while (extra_argv[extra_argc]) extra_argc++;
4837         }
4838
4839         if (extra_argc < 1) {
4840                 usage();
4841         }
4842
4843         if (options.maxruntime == 0) {
4844                 const char *ctdb_timeout;
4845                 ctdb_timeout = getenv("CTDB_TIMEOUT");
4846                 if (ctdb_timeout != NULL) {
4847                         options.maxruntime = strtoul(ctdb_timeout, NULL, 0);
4848                 } else {
4849                         /* default timeout is 120 seconds */
4850                         options.maxruntime = 120;
4851                 }
4852         }
4853
4854         signal(SIGALRM, ctdb_alarm);
4855         alarm(options.maxruntime);
4856
4857         /* setup the node number to contact */
4858         if (nodestring != NULL) {
4859                 if (strcmp(nodestring, "all") == 0) {
4860                         options.pnn = CTDB_BROADCAST_ALL;
4861                 } else {
4862                         options.pnn = strtoul(nodestring, NULL, 0);
4863                 }
4864         }
4865
4866         control = extra_argv[0];
4867
4868         ev = event_context_init(NULL);
4869         if (!ev) {
4870                 DEBUG(DEBUG_ERR, ("Failed to initialize event system\n"));
4871                 exit(1);
4872         }
4873         tevent_loop_allow_nesting(ev);
4874
4875         for (i=0;i<ARRAY_SIZE(ctdb_commands);i++) {
4876                 if (strcmp(control, ctdb_commands[i].name) == 0) {
4877                         int j;
4878
4879                         if (ctdb_commands[i].without_daemon == true) {
4880                                 close(2);
4881                         }
4882
4883                         if (ctdb_commands[i].without_daemon == false) {
4884                                 const char *socket_name;
4885
4886                                 /* initialise ctdb */
4887                                 ctdb = ctdb_cmdline_client(ev);
4888
4889                                 if (ctdb == NULL) {
4890                                         DEBUG(DEBUG_ERR, ("Failed to init ctdb\n"));
4891                                         exit(1);
4892                                 }
4893
4894                                 /* initialize a libctdb connection as well */
4895                                 socket_name = ctdb_get_socketname(ctdb);
4896                                 ctdb_connection = ctdb_connect(socket_name,
4897                                                        ctdb_log_file, stderr);
4898                                 if (ctdb_connection == NULL) {
4899                                         fprintf(stderr, "Failed to connect to daemon from libctdb\n");
4900                                         exit(1);
4901                                 }
4902
4903                                 /* verify the node exists */
4904                                 verify_node(ctdb);
4905
4906                                 if (options.pnn == CTDB_CURRENT_NODE) {
4907                                         int pnn;
4908                                         pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);
4909                                         if (pnn == -1) {
4910                                                 return -1;
4911                                         }
4912                                         options.pnn = pnn;
4913                                 }
4914                         }
4915
4916                         if (ctdb_commands[i].auto_all &&
4917                             options.pnn == CTDB_BROADCAST_ALL) {
4918                                 uint32_t *nodes;
4919                                 uint32_t num_nodes;
4920                                 ret = 0;
4921
4922                                 nodes = ctdb_get_connected_nodes(ctdb, TIMELIMIT(), ctdb, &num_nodes);
4923                                 CTDB_NO_MEMORY(ctdb, nodes);
4924
4925                                 for (j=0;j<num_nodes;j++) {
4926                                         options.pnn = nodes[j];
4927                                         ret |= ctdb_commands[i].fn(ctdb, extra_argc-1, extra_argv+1);
4928                                 }
4929                                 talloc_free(nodes);
4930                         } else {
4931                                 ret = ctdb_commands[i].fn(ctdb, extra_argc-1, extra_argv+1);
4932                         }
4933                         break;
4934                 }
4935         }
4936
4937         if (i == ARRAY_SIZE(ctdb_commands)) {
4938                 DEBUG(DEBUG_ERR, ("Unknown control '%s'\n", control));
4939                 exit(1);
4940         }
4941
4942         return ret;
4943 }