server/ctdb_recoverd.c

   1 /*
   2    ctdb recovery daemon
   3
   4    Copyright (C) Ronnie Sahlberg  2007
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "includes.h"
  21 #include "lib/tevent/tevent.h"
  22 #include "system/filesys.h"
  23 #include "system/time.h"
  24 #include "system/network.h"
  25 #include "system/wait.h"
  26 #include "popt.h"
  27 #include "cmdline.h"
  28 #include "../include/ctdb_client.h"
  29 #include "../include/ctdb_private.h"
  30 #include "db_wrap.h"
  31 #include "dlinklist.h"
  32
  33
  34 /* list of "ctdb ipreallocate" processes to call back when we have
  35    finished the takeover run.
  36 */
  37 struct ip_reallocate_list {
  38         struct ip_reallocate_list *next;
  39         struct rd_memdump_reply *rd;
  40 };
  41
  42 struct ctdb_banning_state {
  43         uint32_t count;
  44         struct timeval last_reported_time;
  45 };
  46
  47 /*
  48   private state of recovery daemon
  49  */
  50 struct ctdb_recoverd {
  51         struct ctdb_context *ctdb;
  52         uint32_t recmaster;
  53         uint32_t num_active;
  54         uint32_t num_connected;
  55         uint32_t last_culprit_node;
  56         struct ctdb_node_map *nodemap;
  57         struct timeval priority_time;
  58         bool need_takeover_run;
  59         bool need_recovery;
  60         uint32_t node_flags;
  61         struct timed_event *send_election_te;
  62         struct timed_event *election_timeout;
  63         struct vacuum_info *vacuum_info;
  64         TALLOC_CTX *ip_reallocate_ctx;
  65         struct ip_reallocate_list *reallocate_callers;
  66         TALLOC_CTX *ip_check_disable_ctx;
  67         struct ctdb_control_get_ifaces *ifaces;
  68 };
  69
  70 #define CONTROL_TIMEOUT() timeval_current_ofs(ctdb->tunable.recover_timeout, 0)
  71 #define MONITOR_TIMEOUT() timeval_current_ofs(ctdb->tunable.recover_interval, 0)
  72
  73 static void ctdb_restart_recd(struct event_context *ev, struct timed_event *te, struct timeval t, void *private_data);
  74
  75 /*
  76   ban a node for a period of time
  77  */
  78 static void ctdb_ban_node(struct ctdb_recoverd *rec, uint32_t pnn, uint32_t ban_time)
  79 {
  80         int ret;
  81         struct ctdb_context *ctdb = rec->ctdb;
  82         struct ctdb_ban_time bantime;
  83
  84         DEBUG(DEBUG_NOTICE,("Banning node %u for %u seconds\n", pnn, ban_time));
  85
  86         if (!ctdb_validate_pnn(ctdb, pnn)) {
  87                 DEBUG(DEBUG_ERR,("Bad pnn %u in ctdb_ban_node\n", pnn));
  88                 return;
  89         }
  90
  91         bantime.pnn  = pnn;
  92         bantime.time = ban_time;
  93
  94         ret = ctdb_ctrl_set_ban(ctdb, CONTROL_TIMEOUT(), pnn, &bantime);
  95         if (ret != 0) {
  96                 DEBUG(DEBUG_ERR,(__location__ " Failed to ban node %d\n", pnn));
  97                 return;
  98         }
  99
 100 }
 101
 102 enum monitor_result { MONITOR_OK, MONITOR_RECOVERY_NEEDED, MONITOR_ELECTION_NEEDED, MONITOR_FAILED};
 103
 104
 105 /*
 106   run the "recovered" eventscript on all nodes
 107  */
 108 static int run_recovered_eventscript(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap, const char *caller)
 109 {
 110         TALLOC_CTX *tmp_ctx;
 111         uint32_t *nodes;
 112
 113         tmp_ctx = talloc_new(ctdb);
 114         CTDB_NO_MEMORY(ctdb, tmp_ctx);
 115
 116         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
 117         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_END_RECOVERY,
 118                                         nodes, 0,
 119                                         CONTROL_TIMEOUT(), false, tdb_null,
 120                                         NULL, NULL,
 121                                         NULL) != 0) {
 122                 DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'recovered' event when called from %s\n", caller));
 123
 124                 talloc_free(tmp_ctx);
 125                 return -1;
 126         }
 127
 128         talloc_free(tmp_ctx);
 129         return 0;
 130 }
 131
 132 /*
 133   remember the trouble maker
 134  */
 135 static void ctdb_set_culprit_count(struct ctdb_recoverd *rec, uint32_t culprit, uint32_t count)
 136 {
 137         struct ctdb_context *ctdb = talloc_get_type(rec->ctdb, struct ctdb_context);
 138         struct ctdb_banning_state *ban_state;
 139
 140         if (culprit > ctdb->num_nodes) {
 141                 DEBUG(DEBUG_ERR,("Trying to set culprit %d but num_nodes is %d\n", culprit, ctdb->num_nodes));
 142                 return;
 143         }
 144
 145         if (ctdb->nodes[culprit]->ban_state == NULL) {
 146                 ctdb->nodes[culprit]->ban_state = talloc_zero(ctdb->nodes[culprit], struct ctdb_banning_state);
 147                 CTDB_NO_MEMORY_VOID(ctdb, ctdb->nodes[culprit]->ban_state);
 148
 149
 150         }
 151         ban_state = ctdb->nodes[culprit]->ban_state;
 152         if (timeval_elapsed(&ban_state->last_reported_time) > ctdb->tunable.recovery_grace_period) {
 153                 /* this was the first time in a long while this node
 154                    misbehaved so we will forgive any old transgressions.
 155                 */
 156                 ban_state->count = 0;
 157         }
 158
 159         ban_state->count += count;
 160         ban_state->last_reported_time = timeval_current();
 161         rec->last_culprit_node = culprit;
 162 }
 163
 164 /*
 165   remember the trouble maker
 166  */
 167 static void ctdb_set_culprit(struct ctdb_recoverd *rec, uint32_t culprit)
 168 {
 169         ctdb_set_culprit_count(rec, culprit, 1);
 170 }
 171
 172
 173 /* this callback is called for every node that failed to execute the
 174    start recovery event
 175 */
 176 static void startrecovery_fail_callback(struct ctdb_context *ctdb, uint32_t node_pnn, int32_t res, TDB_DATA outdata, void *callback_data)
 177 {
 178         struct ctdb_recoverd *rec = talloc_get_type(callback_data, struct ctdb_recoverd);
 179
 180         DEBUG(DEBUG_ERR, (__location__ " Node %u failed the startrecovery event. Setting it as recovery fail culprit\n", node_pnn));
 181
 182         ctdb_set_culprit(rec, node_pnn);
 183 }
 184
 185 /*
 186   run the "startrecovery" eventscript on all nodes
 187  */
 188 static int run_startrecovery_eventscript(struct ctdb_recoverd *rec, struct ctdb_node_map *nodemap)
 189 {
 190         TALLOC_CTX *tmp_ctx;
 191         uint32_t *nodes;
 192         struct ctdb_context *ctdb = rec->ctdb;
 193
 194         tmp_ctx = talloc_new(ctdb);
 195         CTDB_NO_MEMORY(ctdb, tmp_ctx);
 196
 197         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
 198         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_START_RECOVERY,
 199                                         nodes, 0,
 200                                         CONTROL_TIMEOUT(), false, tdb_null,
 201                                         NULL,
 202                                         startrecovery_fail_callback,
 203                                         rec) != 0) {
 204                 DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'startrecovery' event. Recovery failed.\n"));
 205                 talloc_free(tmp_ctx);
 206                 return -1;
 207         }
 208
 209         talloc_free(tmp_ctx);
 210         return 0;
 211 }
 212
 213 static void async_getcap_callback(struct ctdb_context *ctdb, uint32_t node_pnn, int32_t res, TDB_DATA outdata, void *callback_data)
 214 {
 215         if ( (outdata.dsize != sizeof(uint32_t)) || (outdata.dptr == NULL) ) {
 216                 DEBUG(DEBUG_ERR, (__location__ " Invalid length/pointer for getcap callback : %u %p\n",  (unsigned)outdata.dsize, outdata.dptr));
 217                 return;
 218         }
 219         if (node_pnn < ctdb->num_nodes) {
 220                 ctdb->nodes[node_pnn]->capabilities = *((uint32_t *)outdata.dptr);
 221         }
 222 }
 223
 224 /*
 225   update the node capabilities for all connected nodes
 226  */
 227 static int update_capabilities(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap)
 228 {
 229         uint32_t *nodes;
 230         TALLOC_CTX *tmp_ctx;
 231
 232         tmp_ctx = talloc_new(ctdb);
 233         CTDB_NO_MEMORY(ctdb, tmp_ctx);
 234
 235         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
 236         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_GET_CAPABILITIES,
 237                                         nodes, 0,
 238                                         CONTROL_TIMEOUT(),
 239                                         false, tdb_null,
 240                                         async_getcap_callback, NULL,
 241                                         NULL) != 0) {
 242                 DEBUG(DEBUG_ERR, (__location__ " Failed to read node capabilities.\n"));
 243                 talloc_free(tmp_ctx);
 244                 return -1;
 245         }
 246
 247         talloc_free(tmp_ctx);
 248         return 0;
 249 }
 250
 251 static void set_recmode_fail_callback(struct ctdb_context *ctdb, uint32_t node_pnn, int32_t res, TDB_DATA outdata, void *callback_data)
 252 {
 253         struct ctdb_recoverd *rec = talloc_get_type(callback_data, struct ctdb_recoverd);
 254
 255         DEBUG(DEBUG_ERR,("Failed to freeze node %u during recovery. Set it as ban culprit for %d credits\n", node_pnn, rec->nodemap->num));
 256         ctdb_set_culprit_count(rec, node_pnn, rec->nodemap->num);
 257 }
 258
 259 static void transaction_start_fail_callback(struct ctdb_context *ctdb, uint32_t node_pnn, int32_t res, TDB_DATA outdata, void *callback_data)
 260 {
 261         struct ctdb_recoverd *rec = talloc_get_type(callback_data, struct ctdb_recoverd);
 262
 263         DEBUG(DEBUG_ERR,("Failed to start recovery transaction on node %u. Set it as ban culprit for %d credits\n", node_pnn, rec->nodemap->num));
 264         ctdb_set_culprit_count(rec, node_pnn, rec->nodemap->num);
 265 }
 266
 267 /*
 268   change recovery mode on all nodes
 269  */
 270 static int set_recovery_mode(struct ctdb_context *ctdb, struct ctdb_recoverd *rec, struct ctdb_node_map *nodemap, uint32_t rec_mode)
 271 {
 272         TDB_DATA data;
 273         uint32_t *nodes;
 274         TALLOC_CTX *tmp_ctx;
 275
 276         tmp_ctx = talloc_new(ctdb);
 277         CTDB_NO_MEMORY(ctdb, tmp_ctx);
 278
 279         /* freeze all nodes */
 280         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
 281         if (rec_mode == CTDB_RECOVERY_ACTIVE) {
 282                 int i;
 283
 284                 for (i=1; i<=NUM_DB_PRIORITIES; i++) {
 285                         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_FREEZE,
 286                                                 nodes, i,
 287                                                 CONTROL_TIMEOUT(),
 288                                                 false, tdb_null,
 289                                                 NULL,
 290                                                 set_recmode_fail_callback,
 291                                                 rec) != 0) {
 292                                 DEBUG(DEBUG_ERR, (__location__ " Unable to freeze nodes. Recovery failed.\n"));
 293                                 talloc_free(tmp_ctx);
 294                                 return -1;
 295                         }
 296                 }
 297         }
 298
 299
 300         data.dsize = sizeof(uint32_t);
 301         data.dptr = (unsigned char *)&rec_mode;
 302
 303         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_SET_RECMODE,
 304                                         nodes, 0,
 305                                         CONTROL_TIMEOUT(),
 306                                         false, data,
 307                                         NULL, NULL,
 308                                         NULL) != 0) {
 309                 DEBUG(DEBUG_ERR, (__location__ " Unable to set recovery mode. Recovery failed.\n"));
 310                 talloc_free(tmp_ctx);
 311                 return -1;
 312         }
 313
 314         talloc_free(tmp_ctx);
 315         return 0;
 316 }
 317
 318 /*
 319   change recovery master on all node
 320  */
 321 static int set_recovery_master(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap, uint32_t pnn)
 322 {
 323         TDB_DATA data;
 324         TALLOC_CTX *tmp_ctx;
 325         uint32_t *nodes;
 326
 327         tmp_ctx = talloc_new(ctdb);
 328         CTDB_NO_MEMORY(ctdb, tmp_ctx);
 329
 330         data.dsize = sizeof(uint32_t);
 331         data.dptr = (unsigned char *)&pnn;
 332
 333         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
 334         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_SET_RECMASTER,
 335                                         nodes, 0,
 336                                         CONTROL_TIMEOUT(), false, data,
 337                                         NULL, NULL,
 338                                         NULL) != 0) {
 339                 DEBUG(DEBUG_ERR, (__location__ " Unable to set recmaster. Recovery failed.\n"));
 340                 talloc_free(tmp_ctx);
 341                 return -1;
 342         }
 343
 344         talloc_free(tmp_ctx);
 345         return 0;
 346 }
 347
 348 /* update all remote nodes to use the same db priority that we have
 349    this can fail if the remove node has not yet been upgraded to
 350    support this function, so we always return success and never fail
 351    a recovery if this call fails.
 352 */
 353 static int update_db_priority_on_remote_nodes(struct ctdb_context *ctdb,
 354         struct ctdb_node_map *nodemap,
 355         uint32_t pnn, struct ctdb_dbid_map *dbmap, TALLOC_CTX *mem_ctx)
 356 {
 357         int db;
 358         uint32_t *nodes;
 359
 360         nodes = list_of_active_nodes(ctdb, nodemap, mem_ctx, true);
 361
 362         /* step through all local databases */
 363         for (db=0; db<dbmap->num;db++) {
 364                 TDB_DATA data;
 365                 struct ctdb_db_priority db_prio;
 366                 int ret;
 367
 368                 db_prio.db_id     = dbmap->dbs[db].dbid;
 369                 ret = ctdb_ctrl_get_db_priority(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, dbmap->dbs[db].dbid, &db_prio.priority);
 370                 if (ret != 0) {
 371                         DEBUG(DEBUG_ERR,(__location__ " Failed to read database priority from local node for db 0x%08x\n", dbmap->dbs[db].dbid));
 372                         continue;
 373                 }
 374
 375                 DEBUG(DEBUG_INFO,("Update DB priority for db 0x%08x to %u\n", dbmap->dbs[db].dbid, db_prio.priority));
 376
 377                 data.dptr  = (uint8_t *)&db_prio;
 378                 data.dsize = sizeof(db_prio);
 379
 380                 if (ctdb_client_async_control(ctdb,
 381                                         CTDB_CONTROL_SET_DB_PRIORITY,
 382                                         nodes, 0,
 383                                         CONTROL_TIMEOUT(), false, data,
 384                                         NULL, NULL,
 385                                         NULL) != 0) {
 386                         DEBUG(DEBUG_ERR,(__location__ " Failed to set DB priority for 0x%08x\n", db_prio.db_id));
 387                 }
 388         }
 389
 390         return 0;
 391 }
 392
 393 /*
 394   ensure all other nodes have attached to any databases that we have
 395  */
 396 static int create_missing_remote_databases(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap,
 397                                            uint32_t pnn, struct ctdb_dbid_map *dbmap, TALLOC_CTX *mem_ctx)
 398 {
 399         int i, j, db, ret;
 400         struct ctdb_dbid_map *remote_dbmap;
 401
 402         /* verify that all other nodes have all our databases */
 403         for (j=0; j<nodemap->num; j++) {
 404                 /* we dont need to ourself ourselves */
 405                 if (nodemap->nodes[j].pnn == pnn) {
 406                         continue;
 407                 }
 408                 /* dont check nodes that are unavailable */
 409                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
 410                         continue;
 411                 }
 412
 413                 ret = ctdb_ctrl_getdbmap(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn,
 414                                          mem_ctx, &remote_dbmap);
 415                 if (ret != 0) {
 416                         DEBUG(DEBUG_ERR, (__location__ " Unable to get dbids from node %u\n", pnn));
 417                         return -1;
 418                 }
 419
 420                 /* step through all local databases */
 421                 for (db=0; db<dbmap->num;db++) {
 422                         const char *name;
 423
 424
 425                         for (i=0;i<remote_dbmap->num;i++) {
 426                                 if (dbmap->dbs[db].dbid == remote_dbmap->dbs[i].dbid) {
 427                                         break;
 428                                 }
 429                         }
 430                         /* the remote node already have this database */
 431                         if (i!=remote_dbmap->num) {
 432                                 continue;
 433                         }
 434                         /* ok so we need to create this database */
 435                         ctdb_ctrl_getdbname(ctdb, CONTROL_TIMEOUT(), pnn, dbmap->dbs[db].dbid,
 436                                             mem_ctx, &name);
 437                         if (ret != 0) {
 438                                 DEBUG(DEBUG_ERR, (__location__ " Unable to get dbname from node %u\n", pnn));
 439                                 return -1;
 440                         }
 441                         ctdb_ctrl_createdb(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn,
 442                                            mem_ctx, name,
 443                                            dbmap->dbs[db].flags & CTDB_DB_FLAGS_PERSISTENT);
 444                         if (ret != 0) {
 445                                 DEBUG(DEBUG_ERR, (__location__ " Unable to create remote db:%s\n", name));
 446                                 return -1;
 447                         }
 448                 }
 449         }
 450
 451         return 0;
 452 }
 453
 454
 455 /*
 456   ensure we are attached to any databases that anyone else is attached to
 457  */
 458 static int create_missing_local_databases(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap,
 459                                           uint32_t pnn, struct ctdb_dbid_map **dbmap, TALLOC_CTX *mem_ctx)
 460 {
 461         int i, j, db, ret;
 462         struct ctdb_dbid_map *remote_dbmap;
 463
 464         /* verify that we have all database any other node has */
 465         for (j=0; j<nodemap->num; j++) {
 466                 /* we dont need to ourself ourselves */
 467                 if (nodemap->nodes[j].pnn == pnn) {
 468                         continue;
 469                 }
 470                 /* dont check nodes that are unavailable */
 471                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
 472                         continue;
 473                 }
 474
 475                 ret = ctdb_ctrl_getdbmap(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn,
 476                                          mem_ctx, &remote_dbmap);
 477                 if (ret != 0) {
 478                         DEBUG(DEBUG_ERR, (__location__ " Unable to get dbids from node %u\n", pnn));
 479                         return -1;
 480                 }
 481
 482                 /* step through all databases on the remote node */
 483                 for (db=0; db<remote_dbmap->num;db++) {
 484                         const char *name;
 485
 486                         for (i=0;i<(*dbmap)->num;i++) {
 487                                 if (remote_dbmap->dbs[db].dbid == (*dbmap)->dbs[i].dbid) {
 488                                         break;
 489                                 }
 490                         }
 491                         /* we already have this db locally */
 492                         if (i!=(*dbmap)->num) {
 493                                 continue;
 494                         }
 495                         /* ok so we need to create this database and
 496                            rebuild dbmap
 497                          */
 498                         ctdb_ctrl_getdbname(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn,
 499                                             remote_dbmap->dbs[db].dbid, mem_ctx, &name);
 500                         if (ret != 0) {
 501                                 DEBUG(DEBUG_ERR, (__location__ " Unable to get dbname from node %u\n",
 502                                           nodemap->nodes[j].pnn));
 503                                 return -1;
 504                         }
 505                         ctdb_ctrl_createdb(ctdb, CONTROL_TIMEOUT(), pnn, mem_ctx, name,
 506                                            remote_dbmap->dbs[db].flags & CTDB_DB_FLAGS_PERSISTENT);
 507                         if (ret != 0) {
 508                                 DEBUG(DEBUG_ERR, (__location__ " Unable to create local db:%s\n", name));
 509                                 return -1;
 510                         }
 511                         ret = ctdb_ctrl_getdbmap(ctdb, CONTROL_TIMEOUT(), pnn, mem_ctx, dbmap);
 512                         if (ret != 0) {
 513                                 DEBUG(DEBUG_ERR, (__location__ " Unable to reread dbmap on node %u\n", pnn));
 514                                 return -1;
 515                         }
 516                 }
 517         }
 518
 519         return 0;
 520 }
 521
 522
 523 /*
 524   pull the remote database contents from one node into the recdb
 525  */
 526 static int pull_one_remote_database(struct ctdb_context *ctdb, uint32_t srcnode,
 527                                     struct tdb_wrap *recdb, uint32_t dbid,
 528                                     bool persistent)
 529 {
 530         int ret;
 531         TDB_DATA outdata;
 532         struct ctdb_marshall_buffer *reply;
 533         struct ctdb_rec_data *rec;
 534         int i;
 535         TALLOC_CTX *tmp_ctx = talloc_new(recdb);
 536
 537         ret = ctdb_ctrl_pulldb(ctdb, srcnode, dbid, CTDB_LMASTER_ANY, tmp_ctx,
 538                                CONTROL_TIMEOUT(), &outdata);
 539         if (ret != 0) {
 540                 DEBUG(DEBUG_ERR,(__location__ " Unable to copy db from node %u\n", srcnode));
 541                 talloc_free(tmp_ctx);
 542                 return -1;
 543         }
 544
 545         reply = (struct ctdb_marshall_buffer *)outdata.dptr;
 546
 547         if (outdata.dsize < offsetof(struct ctdb_marshall_buffer, data)) {
 548                 DEBUG(DEBUG_ERR,(__location__ " invalid data in pulldb reply\n"));
 549                 talloc_free(tmp_ctx);
 550                 return -1;
 551         }
 552
 553         rec = (struct ctdb_rec_data *)&reply->data[0];
 554
 555         for (i=0;
 556              i<reply->count;
 557              rec = (struct ctdb_rec_data *)(rec->length + (uint8_t *)rec), i++) {
 558                 TDB_DATA key, data;
 559                 struct ctdb_ltdb_header *hdr;
 560                 TDB_DATA existing;
 561
 562                 key.dptr = &rec->data[0];
 563                 key.dsize = rec->keylen;
 564                 data.dptr = &rec->data[key.dsize];
 565                 data.dsize = rec->datalen;
 566
 567                 hdr = (struct ctdb_ltdb_header *)data.dptr;
 568
 569                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
 570                         DEBUG(DEBUG_CRIT,(__location__ " bad ltdb record\n"));
 571                         talloc_free(tmp_ctx);
 572                         return -1;
 573                 }
 574
 575                 /* fetch the existing record, if any */
 576                 existing = tdb_fetch(recdb->tdb, key);
 577
 578                 if (existing.dptr != NULL) {
 579                         struct ctdb_ltdb_header header;
 580                         if (existing.dsize < sizeof(struct ctdb_ltdb_header)) {
 581                                 DEBUG(DEBUG_CRIT,(__location__ " Bad record size %u from node %u\n",
 582                                          (unsigned)existing.dsize, srcnode));
 583                                 free(existing.dptr);
 584                                 talloc_free(tmp_ctx);
 585                                 return -1;
 586                         }
 587                         header = *(struct ctdb_ltdb_header *)existing.dptr;
 588                         free(existing.dptr);
 589                         if (!(header.rsn < hdr->rsn ||
 590                               (header.dmaster != ctdb->recovery_master && header.rsn == hdr->rsn))) {
 591                                 continue;
 592                         }
 593                 }
 594
 595                 if (tdb_store(recdb->tdb, key, data, TDB_REPLACE) != 0) {
 596                         DEBUG(DEBUG_CRIT,(__location__ " Failed to store record\n"));
 597                         talloc_free(tmp_ctx);
 598                         return -1;
 599                 }
 600         }
 601
 602         talloc_free(tmp_ctx);
 603
 604         return 0;
 605 }
 606
 607 /*
 608   pull all the remote database contents into the recdb
 609  */
 610 static int pull_remote_database(struct ctdb_context *ctdb,
 611                                 struct ctdb_recoverd *rec,
 612                                 struct ctdb_node_map *nodemap,
 613                                 struct tdb_wrap *recdb, uint32_t dbid,
 614                                 bool persistent)
 615 {
 616         int j;
 617
 618         /* pull all records from all other nodes across onto this node
 619            (this merges based on rsn)
 620         */
 621         for (j=0; j<nodemap->num; j++) {
 622                 /* dont merge from nodes that are unavailable */
 623                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
 624                         continue;
 625                 }
 626                 if (pull_one_remote_database(ctdb, nodemap->nodes[j].pnn, recdb, dbid, persistent) != 0) {
 627                         DEBUG(DEBUG_ERR,(__location__ " Failed to pull remote database from node %u\n",
 628                                  nodemap->nodes[j].pnn));
 629                         ctdb_set_culprit_count(rec, nodemap->nodes[j].pnn, nodemap->num);
 630                         return -1;
 631                 }
 632         }
 633
 634         return 0;
 635 }
 636
 637
 638 /*
 639   update flags on all active nodes
 640  */
 641 static int update_flags_on_all_nodes(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap, uint32_t pnn, uint32_t flags)
 642 {
 643         int ret;
 644
 645         ret = ctdb_ctrl_modflags(ctdb, CONTROL_TIMEOUT(), pnn, flags, ~flags);
 646                 if (ret != 0) {
 647                 DEBUG(DEBUG_ERR, (__location__ " Unable to update nodeflags on remote nodes\n"));
 648                 return -1;
 649         }
 650
 651         return 0;
 652 }
 653
 654 /*
 655   ensure all nodes have the same vnnmap we do
 656  */
 657 static int update_vnnmap_on_all_nodes(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap,
 658                                       uint32_t pnn, struct ctdb_vnn_map *vnnmap, TALLOC_CTX *mem_ctx)
 659 {
 660         int j, ret;
 661
 662         /* push the new vnn map out to all the nodes */
 663         for (j=0; j<nodemap->num; j++) {
 664                 /* dont push to nodes that are unavailable */
 665                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
 666                         continue;
 667                 }
 668
 669                 ret = ctdb_ctrl_setvnnmap(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn, mem_ctx, vnnmap);
 670                 if (ret != 0) {
 671                         DEBUG(DEBUG_ERR, (__location__ " Unable to set vnnmap for node %u\n", pnn));
 672                         return -1;
 673                 }
 674         }
 675
 676         return 0;
 677 }
 678
 679
 680 struct vacuum_info {
 681         struct vacuum_info *next, *prev;
 682         struct ctdb_recoverd *rec;
 683         uint32_t srcnode;
 684         struct ctdb_db_context *ctdb_db;
 685         struct ctdb_marshall_buffer *recs;
 686         struct ctdb_rec_data *r;
 687 };
 688
 689 static void vacuum_fetch_next(struct vacuum_info *v);
 690
 691 /*
 692   called when a vacuum fetch has completed - just free it and do the next one
 693  */
 694 static void vacuum_fetch_callback(struct ctdb_client_call_state *state)
 695 {
 696         struct vacuum_info *v = talloc_get_type(state->async.private_data, struct vacuum_info);
 697         talloc_free(state);
 698         vacuum_fetch_next(v);
 699 }
 700
 701
 702 /*
 703   process the next element from the vacuum list
 704 */
 705 static void vacuum_fetch_next(struct vacuum_info *v)
 706 {
 707         struct ctdb_call call;
 708         struct ctdb_rec_data *r;
 709
 710         while (v->recs->count) {
 711                 struct ctdb_client_call_state *state;
 712                 TDB_DATA data;
 713                 struct ctdb_ltdb_header *hdr;
 714
 715                 ZERO_STRUCT(call);
 716                 call.call_id = CTDB_NULL_FUNC;
 717                 call.flags = CTDB_IMMEDIATE_MIGRATION;
 718                 call.flags |= CTDB_CALL_FLAG_VACUUM_MIGRATION;
 719
 720                 r = v->r;
 721                 v->r = (struct ctdb_rec_data *)(r->length + (uint8_t *)r);
 722                 v->recs->count--;
 723
 724                 call.key.dptr = &r->data[0];
 725                 call.key.dsize = r->keylen;
 726
 727                 /* ensure we don't block this daemon - just skip a record if we can't get
 728                    the chainlock */
 729                 if (tdb_chainlock_nonblock(v->ctdb_db->ltdb->tdb, call.key) != 0) {
 730                         continue;
 731                 }
 732
 733                 data = tdb_fetch(v->ctdb_db->ltdb->tdb, call.key);
 734                 if (data.dptr == NULL) {
 735                         tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
 736                         continue;
 737                 }
 738
 739                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
 740                         free(data.dptr);
 741                         tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
 742                         continue;
 743                 }
 744
 745                 hdr = (struct ctdb_ltdb_header *)data.dptr;
 746                 if (hdr->dmaster == v->rec->ctdb->pnn) {
 747                         /* its already local */
 748                         free(data.dptr);
 749                         tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
 750                         continue;
 751                 }
 752
 753                 free(data.dptr);
 754
 755                 state = ctdb_call_send(v->ctdb_db, &call);
 756                 tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
 757                 if (state == NULL) {
 758                         DEBUG(DEBUG_ERR,(__location__ " Failed to setup vacuum fetch call\n"));
 759                         talloc_free(v);
 760                         return;
 761                 }
 762                 state->async.fn = vacuum_fetch_callback;
 763                 state->async.private_data = v;
 764                 return;
 765         }
 766
 767         talloc_free(v);
 768 }
 769
 770
 771 /*
 772   destroy a vacuum info structure
 773  */
 774 static int vacuum_info_destructor(struct vacuum_info *v)
 775 {
 776         DLIST_REMOVE(v->rec->vacuum_info, v);
 777         return 0;
 778 }
 779
 780
 781 /*
 782   handler for vacuum fetch
 783 */
 784 static void vacuum_fetch_handler(struct ctdb_context *ctdb, uint64_t srvid,
 785                                  TDB_DATA data, void *private_data)
 786 {
 787         struct ctdb_recoverd *rec = talloc_get_type(private_data, struct ctdb_recoverd);
 788         struct ctdb_marshall_buffer *recs;
 789         int ret, i;
 790         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
 791         const char *name;
 792         struct ctdb_dbid_map *dbmap=NULL;
 793         bool persistent = false;
 794         struct ctdb_db_context *ctdb_db;
 795         struct ctdb_rec_data *r;
 796         uint32_t srcnode;
 797         struct vacuum_info *v;
 798
 799         recs = (struct ctdb_marshall_buffer *)data.dptr;
 800         r = (struct ctdb_rec_data *)&recs->data[0];
 801
 802         if (recs->count == 0) {
 803                 talloc_free(tmp_ctx);
 804                 return;
 805         }
 806
 807         srcnode = r->reqid;
 808
 809         for (v=rec->vacuum_info;v;v=v->next) {
 810                 if (srcnode == v->srcnode && recs->db_id == v->ctdb_db->db_id) {
 811                         /* we're already working on records from this node */
 812                         talloc_free(tmp_ctx);
 813                         return;
 814                 }
 815         }
 816
 817         /* work out if the database is persistent */
 818         ret = ctdb_ctrl_getdbmap(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, tmp_ctx, &dbmap);
 819         if (ret != 0) {
 820                 DEBUG(DEBUG_ERR, (__location__ " Unable to get dbids from local node\n"));
 821                 talloc_free(tmp_ctx);
 822                 return;
 823         }
 824
 825         for (i=0;i<dbmap->num;i++) {
 826                 if (dbmap->dbs[i].dbid == recs->db_id) {
 827                         persistent = dbmap->dbs[i].flags & CTDB_DB_FLAGS_PERSISTENT;
 828                         break;
 829                 }
 830         }
 831         if (i == dbmap->num) {
 832                 DEBUG(DEBUG_ERR, (__location__ " Unable to find db_id 0x%x on local node\n", recs->db_id));
 833                 talloc_free(tmp_ctx);
 834                 return;
 835         }
 836
 837         /* find the name of this database */
 838         if (ctdb_ctrl_getdbname(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, recs->db_id, tmp_ctx, &name) != 0) {
 839                 DEBUG(DEBUG_ERR,(__location__ " Failed to get name of db 0x%x\n", recs->db_id));
 840                 talloc_free(tmp_ctx);
 841                 return;
 842         }
 843
 844         /* attach to it */
 845         ctdb_db = ctdb_attach(ctdb, name, persistent, 0);
 846         if (ctdb_db == NULL) {
 847                 DEBUG(DEBUG_ERR,(__location__ " Failed to attach to database '%s'\n", name));
 848                 talloc_free(tmp_ctx);
 849                 return;
 850         }
 851
 852         v = talloc_zero(rec, struct vacuum_info);
 853         if (v == NULL) {
 854                 DEBUG(DEBUG_CRIT,(__location__ " Out of memory\n"));
 855                 talloc_free(tmp_ctx);
 856                 return;
 857         }
 858
 859         v->rec = rec;
 860         v->srcnode = srcnode;
 861         v->ctdb_db = ctdb_db;
 862         v->recs = talloc_memdup(v, recs, data.dsize);
 863         if (v->recs == NULL) {
 864                 DEBUG(DEBUG_CRIT,(__location__ " Out of memory\n"));
 865                 talloc_free(v);
 866                 talloc_free(tmp_ctx);
 867                 return;
 868         }
 869         v->r =  (struct ctdb_rec_data *)&v->recs->data[0];
 870
 871         DLIST_ADD(rec->vacuum_info, v);
 872
 873         talloc_set_destructor(v, vacuum_info_destructor);
 874
 875         vacuum_fetch_next(v);
 876         talloc_free(tmp_ctx);
 877 }
 878
 879
 880 /*
 881   called when ctdb_wait_timeout should finish
 882  */
 883 static void ctdb_wait_handler(struct event_context *ev, struct timed_event *te,
 884                               struct timeval yt, void *p)
 885 {
 886         uint32_t *timed_out = (uint32_t *)p;
 887         (*timed_out) = 1;
 888 }
 889
 890 /*
 891   wait for a given number of seconds
 892  */
 893 static void ctdb_wait_timeout(struct ctdb_context *ctdb, double secs)
 894 {
 895         uint32_t timed_out = 0;
 896         time_t usecs = (secs - (time_t)secs) * 1000000;
 897         event_add_timed(ctdb->ev, ctdb, timeval_current_ofs(secs, usecs), ctdb_wait_handler, &timed_out);
 898         while (!timed_out) {
 899                 event_loop_once(ctdb->ev);
 900         }
 901 }
 902
 903 /*
 904   called when an election times out (ends)
 905  */
 906 static void ctdb_election_timeout(struct event_context *ev, struct timed_event *te,
 907                                   struct timeval t, void *p)
 908 {
 909         struct ctdb_recoverd *rec = talloc_get_type(p, struct ctdb_recoverd);
 910         rec->election_timeout = NULL;
 911         fast_start = false;
 912
 913         DEBUG(DEBUG_WARNING,(__location__ " Election timed out\n"));
 914 }
 915
 916
 917 /*
 918   wait for an election to finish. It finished election_timeout seconds after
 919   the last election packet is received
 920  */
 921 static void ctdb_wait_election(struct ctdb_recoverd *rec)
 922 {
 923         struct ctdb_context *ctdb = rec->ctdb;
 924         while (rec->election_timeout) {
 925                 event_loop_once(ctdb->ev);
 926         }
 927 }
 928
 929 /*
 930   Update our local flags from all remote connected nodes.
 931   This is only run when we are or we belive we are the recovery master
 932  */
 933 static int update_local_flags(struct ctdb_recoverd *rec, struct ctdb_node_map *nodemap)
 934 {
 935         int j;
 936         struct ctdb_context *ctdb = rec->ctdb;
 937         TALLOC_CTX *mem_ctx = talloc_new(ctdb);
 938
 939         /* get the nodemap for all active remote nodes and verify
 940            they are the same as for this node
 941          */
 942         for (j=0; j<nodemap->num; j++) {
 943                 struct ctdb_node_map *remote_nodemap=NULL;
 944                 int ret;
 945
 946                 if (nodemap->nodes[j].flags & NODE_FLAGS_DISCONNECTED) {
 947                         continue;
 948                 }
 949                 if (nodemap->nodes[j].pnn == ctdb->pnn) {
 950                         continue;
 951                 }
 952
 953                 ret = ctdb_ctrl_getnodemap(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn,
 954                                            mem_ctx, &remote_nodemap);
 955                 if (ret != 0) {
 956                         DEBUG(DEBUG_ERR, (__location__ " Unable to get nodemap from remote node %u\n",
 957                                   nodemap->nodes[j].pnn));
 958                         ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
 959                         talloc_free(mem_ctx);
 960                         return MONITOR_FAILED;
 961                 }
 962                 if (nodemap->nodes[j].flags != remote_nodemap->nodes[j].flags) {
 963                         /* We should tell our daemon about this so it
 964                            updates its flags or else we will log the same
 965                            message again in the next iteration of recovery.
 966                            Since we are the recovery master we can just as
 967                            well update the flags on all nodes.
 968                         */
 969                         ret = ctdb_ctrl_modflags(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn, nodemap->nodes[j].flags, ~nodemap->nodes[j].flags);
 970                         if (ret != 0) {
 971                                 DEBUG(DEBUG_ERR, (__location__ " Unable to update nodeflags on remote nodes\n"));
 972                                 return -1;
 973                         }
 974
 975                         /* Update our local copy of the flags in the recovery
 976                            daemon.
 977                         */
 978                         DEBUG(DEBUG_NOTICE,("Remote node %u had flags 0x%x, local had 0x%x - updating local\n",
 979                                  nodemap->nodes[j].pnn, remote_nodemap->nodes[j].flags,
 980                                  nodemap->nodes[j].flags));
 981                         nodemap->nodes[j].flags = remote_nodemap->nodes[j].flags;
 982                 }
 983                 talloc_free(remote_nodemap);
 984         }
 985         talloc_free(mem_ctx);
 986         return MONITOR_OK;
 987 }
 988
 989
 990 /* Create a new random generation ip.
 991    The generation id can not be the INVALID_GENERATION id
 992 */
 993 static uint32_t new_generation(void)
 994 {
 995         uint32_t generation;
 996
 997         while (1) {
 998                 generation = random();
 999
1000                 if (generation != INVALID_GENERATION) {
1001                         break;
1002                 }
1003         }
1004
1005         return generation;
1006 }
1007
1008
1009 /*
1010   create a temporary working database
1011  */
1012 static struct tdb_wrap *create_recdb(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx)
1013 {
1014         char *name;
1015         struct tdb_wrap *recdb;
1016         unsigned tdb_flags;
1017
1018         /* open up the temporary recovery database */
1019         name = talloc_asprintf(mem_ctx, "%s/recdb.tdb.%u",
1020                                ctdb->db_directory_state,
1021                                ctdb->pnn);
1022         if (name == NULL) {
1023                 return NULL;
1024         }
1025         unlink(name);
1026
1027         tdb_flags = TDB_NOLOCK;
1028         if (ctdb->valgrinding) {
1029                 tdb_flags |= TDB_NOMMAP;
1030         }
1031         tdb_flags |= TDB_DISALLOW_NESTING;
1032
1033         recdb = tdb_wrap_open(mem_ctx, name, ctdb->tunable.database_hash_size,
1034                               tdb_flags, O_RDWR|O_CREAT|O_EXCL, 0600);
1035         if (recdb == NULL) {
1036                 DEBUG(DEBUG_CRIT,(__location__ " Failed to create temp recovery database '%s'\n", name));
1037         }
1038
1039         talloc_free(name);
1040
1041         return recdb;
1042 }
1043
1044
1045 /*
1046    a traverse function for pulling all relevent records from recdb
1047  */
1048 struct recdb_data {
1049         struct ctdb_context *ctdb;
1050         struct ctdb_marshall_buffer *recdata;
1051         uint32_t len;
1052         bool failed;
1053         bool persistent;
1054 };
1055
1056 static int traverse_recdb(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
1057 {
1058         struct recdb_data *params = (struct recdb_data *)p;
1059         struct ctdb_rec_data *rec;
1060         struct ctdb_ltdb_header *hdr;
1061
1062         /* skip empty records */
1063         if (data.dsize <= sizeof(struct ctdb_ltdb_header)) {
1064                 return 0;
1065         }
1066
1067         /* update the dmaster field to point to us */
1068         hdr = (struct ctdb_ltdb_header *)data.dptr;
1069         if (!params->persistent) {
1070                 hdr->dmaster = params->ctdb->pnn;
1071                 hdr->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
1072         }
1073
1074         /* add the record to the blob ready to send to the nodes */
1075         rec = ctdb_marshall_record(params->recdata, 0, key, NULL, data);
1076         if (rec == NULL) {
1077                 params->failed = true;
1078                 return -1;
1079         }
1080         params->recdata = talloc_realloc_size(NULL, params->recdata, rec->length + params->len);
1081         if (params->recdata == NULL) {
1082                 DEBUG(DEBUG_CRIT,(__location__ " Failed to expand recdata to %u (%u records)\n",
1083                          rec->length + params->len, params->recdata->count));
1084                 params->failed = true;
1085                 return -1;
1086         }
1087         params->recdata->count++;
1088         memcpy(params->len+(uint8_t *)params->recdata, rec, rec->length);
1089         params->len += rec->length;
1090         talloc_free(rec);
1091
1092         return 0;
1093 }
1094
1095 /*
1096   push the recdb database out to all nodes
1097  */
1098 static int push_recdb_database(struct ctdb_context *ctdb, uint32_t dbid,
1099                                bool persistent,
1100                                struct tdb_wrap *recdb, struct ctdb_node_map *nodemap)
1101 {
1102         struct recdb_data params;
1103         struct ctdb_marshall_buffer *recdata;
1104         TDB_DATA outdata;
1105         TALLOC_CTX *tmp_ctx;
1106         uint32_t *nodes;
1107
1108         tmp_ctx = talloc_new(ctdb);
1109         CTDB_NO_MEMORY(ctdb, tmp_ctx);
1110
1111         recdata = talloc_zero(recdb, struct ctdb_marshall_buffer);
1112         CTDB_NO_MEMORY(ctdb, recdata);
1113
1114         recdata->db_id = dbid;
1115
1116         params.ctdb = ctdb;
1117         params.recdata = recdata;
1118         params.len = offsetof(struct ctdb_marshall_buffer, data);
1119         params.failed = false;
1120         params.persistent = persistent;
1121
1122         if (tdb_traverse_read(recdb->tdb, traverse_recdb, &params) == -1) {
1123                 DEBUG(DEBUG_ERR,(__location__ " Failed to traverse recdb database\n"));
1124                 talloc_free(params.recdata);
1125                 talloc_free(tmp_ctx);
1126                 return -1;
1127         }
1128
1129         if (params.failed) {
1130                 DEBUG(DEBUG_ERR,(__location__ " Failed to traverse recdb database\n"));
1131                 talloc_free(params.recdata);
1132                 talloc_free(tmp_ctx);
1133                 return -1;
1134         }
1135
1136         recdata = params.recdata;
1137
1138         outdata.dptr = (void *)recdata;
1139         outdata.dsize = params.len;
1140
1141         nodes = list_of_active_nodes(ctdb, nodemap, tmp_ctx, true);
1142         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_PUSH_DB,
1143                                         nodes, 0,
1144                                         CONTROL_TIMEOUT(), false, outdata,
1145                                         NULL, NULL,
1146                                         NULL) != 0) {
1147                 DEBUG(DEBUG_ERR,(__location__ " Failed to push recdb records to nodes for db 0x%x\n", dbid));
1148                 talloc_free(recdata);
1149                 talloc_free(tmp_ctx);
1150                 return -1;
1151         }
1152
1153         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - pushed remote database 0x%x of size %u\n",
1154                   dbid, recdata->count));
1155
1156         talloc_free(recdata);
1157         talloc_free(tmp_ctx);
1158
1159         return 0;
1160 }
1161
1162
1163 /*
1164   go through a full recovery on one database
1165  */
1166 static int recover_database(struct ctdb_recoverd *rec,
1167                             TALLOC_CTX *mem_ctx,
1168                             uint32_t dbid,
1169                             bool persistent,
1170                             uint32_t pnn,
1171                             struct ctdb_node_map *nodemap,
1172                             uint32_t transaction_id)
1173 {
1174         struct tdb_wrap *recdb;
1175         int ret;
1176         struct ctdb_context *ctdb = rec->ctdb;
1177         TDB_DATA data;
1178         struct ctdb_control_wipe_database w;
1179         uint32_t *nodes;
1180
1181         recdb = create_recdb(ctdb, mem_ctx);
1182         if (recdb == NULL) {
1183                 return -1;
1184         }
1185
1186         /* pull all remote databases onto the recdb */
1187         ret = pull_remote_database(ctdb, rec, nodemap, recdb, dbid, persistent);
1188         if (ret != 0) {
1189                 DEBUG(DEBUG_ERR, (__location__ " Unable to pull remote database 0x%x\n", dbid));
1190                 return -1;
1191         }
1192
1193         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - pulled remote database 0x%x\n", dbid));
1194
1195         /* wipe all the remote databases. This is safe as we are in a transaction */
1196         w.db_id = dbid;
1197         w.transaction_id = transaction_id;
1198
1199         data.dptr = (void *)&w;
1200         data.dsize = sizeof(w);
1201
1202         nodes = list_of_active_nodes(ctdb, nodemap, recdb, true);
1203         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_WIPE_DATABASE,
1204                                         nodes, 0,
1205                                         CONTROL_TIMEOUT(), false, data,
1206                                         NULL, NULL,
1207                                         NULL) != 0) {
1208                 DEBUG(DEBUG_ERR, (__location__ " Unable to wipe database. Recovery failed.\n"));
1209                 talloc_free(recdb);
1210                 return -1;
1211         }
1212
1213         /* push out the correct database. This sets the dmaster and skips
1214            the empty records */
1215         ret = push_recdb_database(ctdb, dbid, persistent, recdb, nodemap);
1216         if (ret != 0) {
1217                 talloc_free(recdb);
1218                 return -1;
1219         }
1220
1221         /* all done with this database */
1222         talloc_free(recdb);
1223
1224         return 0;
1225 }
1226
1227 /*
1228   reload the nodes file
1229 */
1230 static void reload_nodes_file(struct ctdb_context *ctdb)
1231 {
1232         ctdb->nodes = NULL;
1233         ctdb_load_nodes_file(ctdb);
1234 }
1235
1236 static int ctdb_reload_remote_public_ips(struct ctdb_context *ctdb,
1237                                          struct ctdb_recoverd *rec,
1238                                          struct ctdb_node_map *nodemap,
1239                                          uint32_t *culprit)
1240 {
1241         int j;
1242         int ret;
1243
1244         if (ctdb->num_nodes != nodemap->num) {
1245                 DEBUG(DEBUG_ERR, (__location__ " ctdb->num_nodes (%d) != nodemap->num (%d) invalid param\n",
1246                                   ctdb->num_nodes, nodemap->num));
1247                 if (culprit) {
1248                         *culprit = ctdb->pnn;
1249                 }
1250                 return -1;
1251         }
1252
1253         for (j=0; j<nodemap->num; j++) {
1254                 /* release any existing data */
1255                 if (ctdb->nodes[j]->known_public_ips) {
1256                         talloc_free(ctdb->nodes[j]->known_public_ips);
1257                         ctdb->nodes[j]->known_public_ips = NULL;
1258                 }
1259                 if (ctdb->nodes[j]->available_public_ips) {
1260                         talloc_free(ctdb->nodes[j]->available_public_ips);
1261                         ctdb->nodes[j]->available_public_ips = NULL;
1262                 }
1263
1264                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
1265                         continue;
1266                 }
1267
1268                 /* grab a new shiny list of public ips from the node */
1269                 ret = ctdb_ctrl_get_public_ips_flags(ctdb,
1270                                         CONTROL_TIMEOUT(),
1271                                         ctdb->nodes[j]->pnn,
1272                                         ctdb->nodes,
1273                                         0,
1274                                         &ctdb->nodes[j]->known_public_ips);
1275                 if (ret != 0) {
1276                         DEBUG(DEBUG_ERR,("Failed to read known public ips from node : %u\n",
1277                                 ctdb->nodes[j]->pnn));
1278                         if (culprit) {
1279                                 *culprit = ctdb->nodes[j]->pnn;
1280                         }
1281                         return -1;
1282                 }
1283
1284                 if (ctdb->tunable.disable_ip_failover == 0) {
1285                         if (rec->ip_check_disable_ctx == NULL) {
1286                                 if (verify_remote_ip_allocation(ctdb, ctdb->nodes[j]->known_public_ips)) {
1287                                         DEBUG(DEBUG_ERR,("Node %d has inconsistent public ip allocation and needs update.\n", ctdb->nodes[j]->pnn));
1288                                         rec->need_takeover_run = true;
1289                                 }
1290                         }
1291                 }
1292
1293                 /* grab a new shiny list of public ips from the node */
1294                 ret = ctdb_ctrl_get_public_ips_flags(ctdb,
1295                                         CONTROL_TIMEOUT(),
1296                                         ctdb->nodes[j]->pnn,
1297                                         ctdb->nodes,
1298                                         CTDB_PUBLIC_IP_FLAGS_ONLY_AVAILABLE,
1299                                         &ctdb->nodes[j]->available_public_ips);
1300                 if (ret != 0) {
1301                         DEBUG(DEBUG_ERR,("Failed to read available public ips from node : %u\n",
1302                                 ctdb->nodes[j]->pnn));
1303                         if (culprit) {
1304                                 *culprit = ctdb->nodes[j]->pnn;
1305                         }
1306                         return -1;
1307                 }
1308         }
1309
1310         return 0;
1311 }
1312
1313 /* when we start a recovery, make sure all nodes use the same reclock file
1314    setting
1315 */
1316 static int sync_recovery_lock_file_across_cluster(struct ctdb_recoverd *rec)
1317 {
1318         struct ctdb_context *ctdb = rec->ctdb;
1319         TALLOC_CTX *tmp_ctx = talloc_new(NULL);
1320         TDB_DATA data;
1321         uint32_t *nodes;
1322
1323         if (ctdb->recovery_lock_file == NULL) {
1324                 data.dptr  = NULL;
1325                 data.dsize = 0;
1326         } else {
1327                 data.dsize = strlen(ctdb->recovery_lock_file) + 1;
1328                 data.dptr  = (uint8_t *)ctdb->recovery_lock_file;
1329         }
1330
1331         nodes = list_of_active_nodes(ctdb, rec->nodemap, tmp_ctx, true);
1332         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_SET_RECLOCK_FILE,
1333                                         nodes, 0,
1334                                         CONTROL_TIMEOUT(),
1335                                         false, data,
1336                                         NULL, NULL,
1337                                         rec) != 0) {
1338                 DEBUG(DEBUG_ERR, (__location__ " Failed to sync reclock file settings\n"));
1339                 talloc_free(tmp_ctx);
1340                 return -1;
1341         }
1342
1343         talloc_free(tmp_ctx);
1344         return 0;
1345 }
1346
1347
1348 /*
1349   we are the recmaster, and recovery is needed - start a recovery run
1350  */
1351 static int do_recovery(struct ctdb_recoverd *rec,
1352                        TALLOC_CTX *mem_ctx, uint32_t pnn,
1353                        struct ctdb_node_map *nodemap, struct ctdb_vnn_map *vnnmap)
1354 {
1355         struct ctdb_context *ctdb = rec->ctdb;
1356         int i, j, ret;
1357         uint32_t generation;
1358         struct ctdb_dbid_map *dbmap;
1359         TDB_DATA data;
1360         uint32_t *nodes;
1361         struct timeval start_time;
1362         uint32_t culprit = (uint32_t)-1;
1363
1364         DEBUG(DEBUG_NOTICE, (__location__ " Starting do_recovery\n"));
1365
1366         /* if recovery fails, force it again */
1367         rec->need_recovery = true;
1368
1369         for (i=0; i<ctdb->num_nodes; i++) {
1370                 struct ctdb_banning_state *ban_state;
1371
1372                 if (ctdb->nodes[i]->ban_state == NULL) {
1373                         continue;
1374                 }
1375                 ban_state = (struct ctdb_banning_state *)ctdb->nodes[i]->ban_state;
1376                 if (ban_state->count < 2*ctdb->num_nodes) {
1377                         continue;
1378                 }
1379                 DEBUG(DEBUG_NOTICE,("Node %u has caused %u recoveries recently - banning it for %u seconds\n",
1380                         ctdb->nodes[i]->pnn, ban_state->count,
1381                         ctdb->tunable.recovery_ban_period));
1382                 ctdb_ban_node(rec, ctdb->nodes[i]->pnn, ctdb->tunable.recovery_ban_period);
1383                 ban_state->count = 0;
1384         }
1385
1386
1387         if (ctdb->tunable.verify_recovery_lock != 0) {
1388                 DEBUG(DEBUG_ERR,("Taking out recovery lock from recovery daemon\n"));
1389                 start_time = timeval_current();
1390                 if (!ctdb_recovery_lock(ctdb, true)) {
1391                         ctdb_set_culprit(rec, pnn);
1392                         DEBUG(DEBUG_ERR,("Unable to get recovery lock - aborting recovery\n"));
1393                         return -1;
1394                 }
1395                 ctdb_ctrl_report_recd_lock_latency(ctdb, CONTROL_TIMEOUT(), timeval_elapsed(&start_time));
1396                 DEBUG(DEBUG_NOTICE,("Recovery lock taken successfully by recovery daemon\n"));
1397         }
1398
1399         DEBUG(DEBUG_NOTICE, (__location__ " Recovery initiated due to problem with node %u\n", rec->last_culprit_node));
1400
1401         /* get a list of all databases */
1402         ret = ctdb_ctrl_getdbmap(ctdb, CONTROL_TIMEOUT(), pnn, mem_ctx, &dbmap);
1403         if (ret != 0) {
1404                 DEBUG(DEBUG_ERR, (__location__ " Unable to get dbids from node :%u\n", pnn));
1405                 return -1;
1406         }
1407
1408         /* we do the db creation before we set the recovery mode, so the freeze happens
1409            on all databases we will be dealing with. */
1410
1411         /* verify that we have all the databases any other node has */
1412         ret = create_missing_local_databases(ctdb, nodemap, pnn, &dbmap, mem_ctx);
1413         if (ret != 0) {
1414                 DEBUG(DEBUG_ERR, (__location__ " Unable to create missing local databases\n"));
1415                 return -1;
1416         }
1417
1418         /* verify that all other nodes have all our databases */
1419         ret = create_missing_remote_databases(ctdb, nodemap, pnn, dbmap, mem_ctx);
1420         if (ret != 0) {
1421                 DEBUG(DEBUG_ERR, (__location__ " Unable to create missing remote databases\n"));
1422                 return -1;
1423         }
1424         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - created remote databases\n"));
1425
1426         /* update the database priority for all remote databases */
1427         ret = update_db_priority_on_remote_nodes(ctdb, nodemap, pnn, dbmap, mem_ctx);
1428         if (ret != 0) {
1429                 DEBUG(DEBUG_ERR, (__location__ " Unable to set db priority on remote nodes\n"));
1430         }
1431         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - updated db priority for all databases\n"));
1432
1433
1434         /* update all other nodes to use the same setting for reclock files
1435            as the local recovery master.
1436         */
1437         sync_recovery_lock_file_across_cluster(rec);
1438
1439         /* set recovery mode to active on all nodes */
1440         ret = set_recovery_mode(ctdb, rec, nodemap, CTDB_RECOVERY_ACTIVE);
1441         if (ret != 0) {
1442                 DEBUG(DEBUG_ERR, (__location__ " Unable to set recovery mode to active on cluster\n"));
1443                 return -1;
1444         }
1445
1446         /* execute the "startrecovery" event script on all nodes */
1447         ret = run_startrecovery_eventscript(rec, nodemap);
1448         if (ret!=0) {
1449                 DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'startrecovery' event on cluster\n"));
1450                 return -1;
1451         }
1452
1453         /*
1454           update all nodes to have the same flags that we have
1455          */
1456         for (i=0;i<nodemap->num;i++) {
1457                 if (nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED) {
1458                         continue;
1459                 }
1460
1461                 ret = update_flags_on_all_nodes(ctdb, nodemap, i, nodemap->nodes[i].flags);
1462                 if (ret != 0) {
1463                         DEBUG(DEBUG_ERR, (__location__ " Unable to update flags on all nodes for node %d\n", i));
1464                         return -1;
1465                 }
1466         }
1467
1468         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - updated flags\n"));
1469
1470         /* pick a new generation number */
1471         generation = new_generation();
1472
1473         /* change the vnnmap on this node to use the new generation
1474            number but not on any other nodes.
1475            this guarantees that if we abort the recovery prematurely
1476            for some reason (a node stops responding?)
1477            that we can just return immediately and we will reenter
1478            recovery shortly again.
1479            I.e. we deliberately leave the cluster with an inconsistent
1480            generation id to allow us to abort recovery at any stage and
1481            just restart it from scratch.
1482          */
1483         vnnmap->generation = generation;
1484         ret = ctdb_ctrl_setvnnmap(ctdb, CONTROL_TIMEOUT(), pnn, mem_ctx, vnnmap);
1485         if (ret != 0) {
1486                 DEBUG(DEBUG_ERR, (__location__ " Unable to set vnnmap for node %u\n", pnn));
1487                 return -1;
1488         }
1489
1490         data.dptr = (void *)&generation;
1491         data.dsize = sizeof(uint32_t);
1492
1493         nodes = list_of_active_nodes(ctdb, nodemap, mem_ctx, true);
1494         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_TRANSACTION_START,
1495                                         nodes, 0,
1496                                         CONTROL_TIMEOUT(), false, data,
1497                                         NULL,
1498                                         transaction_start_fail_callback,
1499                                         rec) != 0) {
1500                 DEBUG(DEBUG_ERR, (__location__ " Unable to start transactions. Recovery failed.\n"));
1501                 if (ctdb_client_async_control(ctdb, CTDB_CONTROL_TRANSACTION_CANCEL,
1502                                         nodes, 0,
1503                                         CONTROL_TIMEOUT(), false, tdb_null,
1504                                         NULL,
1505                                         NULL,
1506                                         NULL) != 0) {
1507                         DEBUG(DEBUG_ERR,("Failed to cancel recovery transaction\n"));
1508                 }
1509                 return -1;
1510         }
1511
1512         DEBUG(DEBUG_NOTICE,(__location__ " started transactions on all nodes\n"));
1513
1514         for (i=0;i<dbmap->num;i++) {
1515                 ret = recover_database(rec, mem_ctx,
1516                                        dbmap->dbs[i].dbid,
1517                                        dbmap->dbs[i].flags & CTDB_DB_FLAGS_PERSISTENT,
1518                                        pnn, nodemap, generation);
1519                 if (ret != 0) {
1520                         DEBUG(DEBUG_ERR, (__location__ " Failed to recover database 0x%x\n", dbmap->dbs[i].dbid));
1521                         return -1;
1522                 }
1523         }
1524
1525         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - starting database commits\n"));
1526
1527         /* commit all the changes */
1528         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_TRANSACTION_COMMIT,
1529                                         nodes, 0,
1530                                         CONTROL_TIMEOUT(), false, data,
1531                                         NULL, NULL,
1532                                         NULL) != 0) {
1533                 DEBUG(DEBUG_ERR, (__location__ " Unable to commit recovery changes. Recovery failed.\n"));
1534                 return -1;
1535         }
1536
1537         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - committed databases\n"));
1538
1539
1540         /* update the capabilities for all nodes */
1541         ret = update_capabilities(ctdb, nodemap);
1542         if (ret!=0) {
1543                 DEBUG(DEBUG_ERR, (__location__ " Unable to update node capabilities.\n"));
1544                 return -1;
1545         }
1546
1547         /* build a new vnn map with all the currently active and
1548            unbanned nodes */
1549         generation = new_generation();
1550         vnnmap = talloc(mem_ctx, struct ctdb_vnn_map);
1551         CTDB_NO_MEMORY(ctdb, vnnmap);
1552         vnnmap->generation = generation;
1553         vnnmap->size = 0;
1554         vnnmap->map = talloc_zero_array(vnnmap, uint32_t, vnnmap->size);
1555         CTDB_NO_MEMORY(ctdb, vnnmap->map);
1556         for (i=j=0;i<nodemap->num;i++) {
1557                 if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
1558                         continue;
1559                 }
1560                 if (!(ctdb->nodes[i]->capabilities & CTDB_CAP_LMASTER)) {
1561                         /* this node can not be an lmaster */
1562                         DEBUG(DEBUG_DEBUG, ("Node %d cant be a LMASTER, skipping it\n", i));
1563                         continue;
1564                 }
1565
1566                 vnnmap->size++;
1567                 vnnmap->map = talloc_realloc(vnnmap, vnnmap->map, uint32_t, vnnmap->size);
1568                 CTDB_NO_MEMORY(ctdb, vnnmap->map);
1569                 vnnmap->map[j++] = nodemap->nodes[i].pnn;
1570
1571         }
1572         if (vnnmap->size == 0) {
1573                 DEBUG(DEBUG_NOTICE, ("No suitable lmasters found. Adding local node (recmaster) anyway.\n"));
1574                 vnnmap->size++;
1575                 vnnmap->map = talloc_realloc(vnnmap, vnnmap->map, uint32_t, vnnmap->size);
1576                 CTDB_NO_MEMORY(ctdb, vnnmap->map);
1577                 vnnmap->map[0] = pnn;
1578         }
1579
1580         /* update to the new vnnmap on all nodes */
1581         ret = update_vnnmap_on_all_nodes(ctdb, nodemap, pnn, vnnmap, mem_ctx);
1582         if (ret != 0) {
1583                 DEBUG(DEBUG_ERR, (__location__ " Unable to update vnnmap on all nodes\n"));
1584                 return -1;
1585         }
1586
1587         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - updated vnnmap\n"));
1588
1589         /* update recmaster to point to us for all nodes */
1590         ret = set_recovery_master(ctdb, nodemap, pnn);
1591         if (ret!=0) {
1592                 DEBUG(DEBUG_ERR, (__location__ " Unable to set recovery master\n"));
1593                 return -1;
1594         }
1595
1596         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - updated recmaster\n"));
1597
1598         /*
1599           update all nodes to have the same flags that we have
1600          */
1601         for (i=0;i<nodemap->num;i++) {
1602                 if (nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED) {
1603                         continue;
1604                 }
1605
1606                 ret = update_flags_on_all_nodes(ctdb, nodemap, i, nodemap->nodes[i].flags);
1607                 if (ret != 0) {
1608                         DEBUG(DEBUG_ERR, (__location__ " Unable to update flags on all nodes for node %d\n", i));
1609                         return -1;
1610                 }
1611         }
1612
1613         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - updated flags\n"));
1614
1615         /* disable recovery mode */
1616         ret = set_recovery_mode(ctdb, rec, nodemap, CTDB_RECOVERY_NORMAL);
1617         if (ret != 0) {
1618                 DEBUG(DEBUG_ERR, (__location__ " Unable to set recovery mode to normal on cluster\n"));
1619                 return -1;
1620         }
1621
1622         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - disabled recovery mode\n"));
1623
1624         /*
1625           tell nodes to takeover their public IPs
1626          */
1627         ret = ctdb_reload_remote_public_ips(ctdb, rec, nodemap, &culprit);
1628         if (ret != 0) {
1629                 DEBUG(DEBUG_ERR,("Failed to read public ips from remote node %d\n",
1630                                  culprit));
1631                 rec->need_takeover_run = true;
1632                 return -1;
1633         }
1634         rec->need_takeover_run = false;
1635         ret = ctdb_takeover_run(ctdb, nodemap);
1636         if (ret != 0) {
1637                 DEBUG(DEBUG_ERR, (__location__ " Unable to setup public takeover addresses. ctdb_takeover_run() failed.\n"));
1638                 rec->need_takeover_run = true;
1639         }
1640
1641         /* execute the "recovered" event script on all nodes */
1642         ret = run_recovered_eventscript(ctdb, nodemap, "do_recovery");
1643         if (ret!=0) {
1644                 DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'recovered' event on cluster. Recovery process failed.\n"));
1645                 return -1;
1646         }
1647
1648         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - finished the recovered event\n"));
1649
1650         /* send a message to all clients telling them that the cluster
1651            has been reconfigured */
1652         ctdb_client_send_message(ctdb, CTDB_BROADCAST_CONNECTED, CTDB_SRVID_RECONFIGURE, tdb_null);
1653
1654         DEBUG(DEBUG_NOTICE, (__location__ " Recovery complete\n"));
1655
1656         rec->need_recovery = false;
1657
1658         /* we managed to complete a full recovery, make sure to forgive
1659            any past sins by the nodes that could now participate in the
1660            recovery.
1661         */
1662         DEBUG(DEBUG_ERR,("Resetting ban count to 0 for all nodes\n"));
1663         for (i=0;i<nodemap->num;i++) {
1664                 struct ctdb_banning_state *ban_state;
1665
1666                 if (nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED) {
1667                         continue;
1668                 }
1669
1670                 ban_state = (struct ctdb_banning_state *)ctdb->nodes[nodemap->nodes[i].pnn]->ban_state;
1671                 if (ban_state == NULL) {
1672                         continue;
1673                 }
1674
1675                 ban_state->count = 0;
1676         }
1677
1678
1679         /* We just finished a recovery successfully.
1680            We now wait for rerecovery_timeout before we allow
1681            another recovery to take place.
1682         */
1683         DEBUG(DEBUG_NOTICE, ("Just finished a recovery. New recoveries will now be supressed for the rerecovery timeout (%d seconds)\n", ctdb->tunable.rerecovery_timeout));
1684         ctdb_wait_timeout(ctdb, ctdb->tunable.rerecovery_timeout);
1685         DEBUG(DEBUG_NOTICE, ("The rerecovery timeout has elapsed. We now allow recoveries to trigger again.\n"));
1686
1687         return 0;
1688 }
1689
1690
1691 /*
1692   elections are won by first checking the number of connected nodes, then
1693   the priority time, then the pnn
1694  */
1695 struct election_message {
1696         uint32_t num_connected;
1697         struct timeval priority_time;
1698         uint32_t pnn;
1699         uint32_t node_flags;
1700 };
1701
1702 /*
1703   form this nodes election data
1704  */
1705 static void ctdb_election_data(struct ctdb_recoverd *rec, struct election_message *em)
1706 {
1707         int ret, i;
1708         struct ctdb_node_map *nodemap;
1709         struct ctdb_context *ctdb = rec->ctdb;
1710
1711         ZERO_STRUCTP(em);
1712
1713         em->pnn = rec->ctdb->pnn;
1714         em->priority_time = rec->priority_time;
1715
1716         ret = ctdb_ctrl_getnodemap(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, rec, &nodemap);
1717         if (ret != 0) {
1718                 DEBUG(DEBUG_ERR,(__location__ " unable to get election data\n"));
1719                 return;
1720         }
1721
1722         rec->node_flags = nodemap->nodes[ctdb->pnn].flags;
1723         em->node_flags = rec->node_flags;
1724
1725         for (i=0;i<nodemap->num;i++) {
1726                 if (!(nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED)) {
1727                         em->num_connected++;
1728                 }
1729         }
1730
1731         /* we shouldnt try to win this election if we cant be a recmaster */
1732         if ((ctdb->capabilities & CTDB_CAP_RECMASTER) == 0) {
1733                 em->num_connected = 0;
1734                 em->priority_time = timeval_current();
1735         }
1736
1737         talloc_free(nodemap);
1738 }
1739
1740 /*
1741   see if the given election data wins
1742  */
1743 static bool ctdb_election_win(struct ctdb_recoverd *rec, struct election_message *em)
1744 {
1745         struct election_message myem;
1746         int cmp = 0;
1747
1748         ctdb_election_data(rec, &myem);
1749
1750         /* we cant win if we dont have the recmaster capability */
1751         if ((rec->ctdb->capabilities & CTDB_CAP_RECMASTER) == 0) {
1752                 return false;
1753         }
1754
1755         /* we cant win if we are banned */
1756         if (rec->node_flags & NODE_FLAGS_BANNED) {
1757                 return false;
1758         }
1759
1760         /* we cant win if we are stopped */
1761         if (rec->node_flags & NODE_FLAGS_STOPPED) {
1762                 return false;
1763         }
1764
1765         /* we will automatically win if the other node is banned */
1766         if (em->node_flags & NODE_FLAGS_BANNED) {
1767                 return true;
1768         }
1769
1770         /* we will automatically win if the other node is banned */
1771         if (em->node_flags & NODE_FLAGS_STOPPED) {
1772                 return true;
1773         }
1774
1775         /* try to use the most connected node */
1776         if (cmp == 0) {
1777                 cmp = (int)myem.num_connected - (int)em->num_connected;
1778         }
1779
1780         /* then the longest running node */
1781         if (cmp == 0) {
1782                 cmp = timeval_compare(&em->priority_time, &myem.priority_time);
1783         }
1784
1785         if (cmp == 0) {
1786                 cmp = (int)myem.pnn - (int)em->pnn;
1787         }
1788
1789         return cmp > 0;
1790 }
1791
1792 /*
1793   send out an election request
1794  */
1795 static int send_election_request(struct ctdb_recoverd *rec, uint32_t pnn, bool update_recmaster)
1796 {
1797         int ret;
1798         TDB_DATA election_data;
1799         struct election_message emsg;
1800         uint64_t srvid;
1801         struct ctdb_context *ctdb = rec->ctdb;
1802
1803         srvid = CTDB_SRVID_RECOVERY;
1804
1805         ctdb_election_data(rec, &emsg);
1806
1807         election_data.dsize = sizeof(struct election_message);
1808         election_data.dptr  = (unsigned char *)&emsg;
1809
1810
1811         /* send an election message to all active nodes */
1812         DEBUG(DEBUG_INFO,(__location__ " Send election request to all active nodes\n"));
1813         ctdb_client_send_message(ctdb, CTDB_BROADCAST_ALL, srvid, election_data);
1814
1815
1816         /* A new node that is already frozen has entered the cluster.
1817            The existing nodes are not frozen and dont need to be frozen
1818            until the election has ended and we start the actual recovery
1819         */
1820         if (update_recmaster == true) {
1821                 /* first we assume we will win the election and set
1822                    recoverymaster to be ourself on the current node
1823                  */
1824                 ret = ctdb_ctrl_setrecmaster(ctdb, CONTROL_TIMEOUT(), pnn, pnn);
1825                 if (ret != 0) {
1826                         DEBUG(DEBUG_ERR, (__location__ " failed to send recmaster election request\n"));
1827                         return -1;
1828                 }
1829         }
1830
1831
1832         return 0;
1833 }
1834
1835 /*
1836   this function will unban all nodes in the cluster
1837 */
1838 static void unban_all_nodes(struct ctdb_context *ctdb)
1839 {
1840         int ret, i;
1841         struct ctdb_node_map *nodemap;
1842         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1843
1844         ret = ctdb_ctrl_getnodemap(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, tmp_ctx, &nodemap);
1845         if (ret != 0) {
1846                 DEBUG(DEBUG_ERR,(__location__ " failed to get nodemap to unban all nodes\n"));
1847                 return;
1848         }
1849
1850         for (i=0;i<nodemap->num;i++) {
1851                 if ( (!(nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED))
1852                   && (nodemap->nodes[i].flags & NODE_FLAGS_BANNED) ) {
1853                         ctdb_ctrl_modflags(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[i].pnn, 0, NODE_FLAGS_BANNED);
1854                 }
1855         }
1856
1857         talloc_free(tmp_ctx);
1858 }
1859
1860
1861 /*
1862   we think we are winning the election - send a broadcast election request
1863  */
1864 static void election_send_request(struct event_context *ev, struct timed_event *te, struct timeval t, void *p)
1865 {
1866         struct ctdb_recoverd *rec = talloc_get_type(p, struct ctdb_recoverd);
1867         int ret;
1868
1869         ret = send_election_request(rec, ctdb_get_pnn(rec->ctdb), false);
1870         if (ret != 0) {
1871                 DEBUG(DEBUG_ERR,("Failed to send election request!\n"));
1872         }
1873
1874         talloc_free(rec->send_election_te);
1875         rec->send_election_te = NULL;
1876 }
1877
1878 /*
1879   handler for memory dumps
1880 */
1881 static void mem_dump_handler(struct ctdb_context *ctdb, uint64_t srvid,
1882                              TDB_DATA data, void *private_data)
1883 {
1884         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
1885         TDB_DATA *dump;
1886         int ret;
1887         struct rd_memdump_reply *rd;
1888
1889         if (data.dsize != sizeof(struct rd_memdump_reply)) {
1890                 DEBUG(DEBUG_ERR, (__location__ " Wrong size of return address.\n"));
1891                 talloc_free(tmp_ctx);
1892                 return;
1893         }
1894         rd = (struct rd_memdump_reply *)data.dptr;
1895
1896         dump = talloc_zero(tmp_ctx, TDB_DATA);
1897         if (dump == NULL) {
1898                 DEBUG(DEBUG_ERR, (__location__ " Failed to allocate memory for memdump\n"));
1899                 talloc_free(tmp_ctx);
1900                 return;
1901         }
1902         ret = ctdb_dump_memory(ctdb, dump);
1903         if (ret != 0) {
1904                 DEBUG(DEBUG_ERR, (__location__ " ctdb_dump_memory() failed\n"));
1905                 talloc_free(tmp_ctx);
1906                 return;
1907         }
1908
1909 DEBUG(DEBUG_ERR, ("recovery master memory dump\n"));
1910
1911         ret = ctdb_client_send_message(ctdb, rd->pnn, rd->srvid, *dump);
1912         if (ret != 0) {
1913                 DEBUG(DEBUG_ERR,("Failed to send rd memdump reply message\n"));
1914                 talloc_free(tmp_ctx);
1915                 return;
1916         }
1917
1918         talloc_free(tmp_ctx);
1919 }
1920
1921 /*
1922   handler for reload_nodes
1923 */
1924 static void reload_nodes_handler(struct ctdb_context *ctdb, uint64_t srvid,
1925                              TDB_DATA data, void *private_data)
1926 {
1927         struct ctdb_recoverd *rec = talloc_get_type(private_data, struct ctdb_recoverd);
1928
1929         DEBUG(DEBUG_ERR, (__location__ " Reload nodes file from recovery daemon\n"));
1930
1931         reload_nodes_file(rec->ctdb);
1932 }
1933
1934
1935 static void reenable_ip_check(struct event_context *ev, struct timed_event *te,
1936                               struct timeval yt, void *p)
1937 {
1938         struct ctdb_recoverd *rec = talloc_get_type(p, struct ctdb_recoverd);
1939
1940         talloc_free(rec->ip_check_disable_ctx);
1941         rec->ip_check_disable_ctx = NULL;
1942 }
1943
1944
1945 static void recd_update_ip_handler(struct ctdb_context *ctdb, uint64_t srvid,
1946                              TDB_DATA data, void *private_data)
1947 {
1948         struct ctdb_recoverd *rec = talloc_get_type(private_data, struct ctdb_recoverd);
1949         struct ctdb_public_ip *ip;
1950
1951         if (rec->recmaster != rec->ctdb->pnn) {
1952                 DEBUG(DEBUG_INFO,("Not recmaster, ignore update ip message\n"));
1953                 return;
1954         }
1955
1956         if (data.dsize != sizeof(struct ctdb_public_ip)) {
1957                 DEBUG(DEBUG_ERR,(__location__ " Incorrect size of recd update ip message. Was %zd but expected %zd bytes\n", data.dsize, sizeof(struct ctdb_public_ip)));
1958                 return;
1959         }
1960
1961         ip = (struct ctdb_public_ip *)data.dptr;
1962
1963         update_ip_assignment_tree(rec->ctdb, ip);
1964 }
1965
1966
1967 static void disable_ip_check_handler(struct ctdb_context *ctdb, uint64_t srvid,
1968                              TDB_DATA data, void *private_data)
1969 {
1970         struct ctdb_recoverd *rec = talloc_get_type(private_data, struct ctdb_recoverd);
1971         uint32_t timeout;
1972
1973         if (rec->ip_check_disable_ctx != NULL) {
1974                 talloc_free(rec->ip_check_disable_ctx);
1975                 rec->ip_check_disable_ctx = NULL;
1976         }
1977
1978         if (data.dsize != sizeof(uint32_t)) {
1979                 DEBUG(DEBUG_ERR,(__location__ " Wrong size for data :%lu "
1980                                  "expexting %lu\n", (long unsigned)data.dsize,
1981                                  (long unsigned)sizeof(uint32_t)));
1982                 return;
1983         }
1984         if (data.dptr == NULL) {
1985                 DEBUG(DEBUG_ERR,(__location__ " No data recaived\n"));
1986                 return;
1987         }
1988
1989         timeout = *((uint32_t *)data.dptr);
1990         DEBUG(DEBUG_NOTICE,("Disabling ip check for %u seconds\n", timeout));
1991
1992         rec->ip_check_disable_ctx = talloc_new(rec);
1993         CTDB_NO_MEMORY_VOID(ctdb, rec->ip_check_disable_ctx);
1994
1995         event_add_timed(ctdb->ev, rec->ip_check_disable_ctx, timeval_current_ofs(timeout, 0), reenable_ip_check, rec);
1996 }
1997
1998
1999 /*
2000   handler for ip reallocate, just add it to the list of callers and
2001   handle this later in the monitor_cluster loop so we do not recurse
2002   with other callers to takeover_run()
2003 */
2004 static void ip_reallocate_handler(struct ctdb_context *ctdb, uint64_t srvid,
2005                              TDB_DATA data, void *private_data)
2006 {
2007         struct ctdb_recoverd *rec = talloc_get_type(private_data, struct ctdb_recoverd);
2008         struct ip_reallocate_list *caller;
2009
2010         if (data.dsize != sizeof(struct rd_memdump_reply)) {
2011                 DEBUG(DEBUG_ERR, (__location__ " Wrong size of return address.\n"));
2012                 return;
2013         }
2014
2015         if (rec->ip_reallocate_ctx == NULL) {
2016                 rec->ip_reallocate_ctx = talloc_new(rec);
2017                 CTDB_NO_MEMORY_FATAL(ctdb, rec->ip_reallocate_ctx);
2018         }
2019
2020         caller = talloc(rec->ip_reallocate_ctx, struct ip_reallocate_list);
2021         CTDB_NO_MEMORY_FATAL(ctdb, caller);
2022
2023         caller->rd   = (struct rd_memdump_reply *)talloc_steal(caller, data.dptr);
2024         caller->next = rec->reallocate_callers;
2025         rec->reallocate_callers = caller;
2026
2027         return;
2028 }
2029
2030 static void process_ipreallocate_requests(struct ctdb_context *ctdb, struct ctdb_recoverd *rec)
2031 {
2032         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
2033         TDB_DATA result;
2034         int32_t ret;
2035         struct ip_reallocate_list *callers;
2036         uint32_t culprit;
2037
2038         DEBUG(DEBUG_INFO, ("recovery master forced ip reallocation\n"));
2039
2040         /* update the list of public ips that a node can handle for
2041            all connected nodes
2042         */
2043         ret = ctdb_reload_remote_public_ips(ctdb, rec, rec->nodemap, &culprit);
2044         if (ret != 0) {
2045                 DEBUG(DEBUG_ERR,("Failed to read public ips from remote node %d\n",
2046                                  culprit));
2047                 rec->need_takeover_run = true;
2048         }
2049         if (ret == 0) {
2050                 ret = ctdb_takeover_run(ctdb, rec->nodemap);
2051                 if (ret != 0) {
2052                         DEBUG(DEBUG_ERR,("Failed to reallocate addresses: ctdb_takeover_run() failed.\n"));
2053                         rec->need_takeover_run = true;
2054                 }
2055         }
2056
2057         result.dsize = sizeof(int32_t);
2058         result.dptr  = (uint8_t *)&ret;
2059
2060         for (callers=rec->reallocate_callers; callers; callers=callers->next) {
2061
2062                 /* Someone that sent srvid==0 does not want a reply */
2063                 if (callers->rd->srvid == 0) {
2064                         continue;
2065                 }
2066                 DEBUG(DEBUG_INFO,("Sending ip reallocate reply message to "
2067                                   "%u:%llu\n", (unsigned)callers->rd->pnn,
2068                                   (unsigned long long)callers->rd->srvid));
2069                 ret = ctdb_client_send_message(ctdb, callers->rd->pnn, callers->rd->srvid, result);
2070                 if (ret != 0) {
2071                         DEBUG(DEBUG_ERR,("Failed to send ip reallocate reply "
2072                                          "message to %u:%llu\n",
2073                                          (unsigned)callers->rd->pnn,
2074                                          (unsigned long long)callers->rd->srvid));
2075                 }
2076         }
2077
2078         talloc_free(tmp_ctx);
2079         talloc_free(rec->ip_reallocate_ctx);
2080         rec->ip_reallocate_ctx = NULL;
2081         rec->reallocate_callers = NULL;
2082
2083 }
2084
2085
2086 /*
2087   handler for recovery master elections
2088 */
2089 static void election_handler(struct ctdb_context *ctdb, uint64_t srvid,
2090                              TDB_DATA data, void *private_data)
2091 {
2092         struct ctdb_recoverd *rec = talloc_get_type(private_data, struct ctdb_recoverd);
2093         int ret;
2094         struct election_message *em = (struct election_message *)data.dptr;
2095         TALLOC_CTX *mem_ctx;
2096
2097         /* we got an election packet - update the timeout for the election */
2098         talloc_free(rec->election_timeout);
2099         rec->election_timeout = event_add_timed(ctdb->ev, ctdb,
2100                                                 fast_start ?
2101                                                 timeval_current_ofs(0, 500000) :
2102                                                 timeval_current_ofs(ctdb->tunable.election_timeout, 0),
2103                                                 ctdb_election_timeout, rec);
2104
2105         mem_ctx = talloc_new(ctdb);
2106
2107         /* someone called an election. check their election data
2108            and if we disagree and we would rather be the elected node,
2109            send a new election message to all other nodes
2110          */
2111         if (ctdb_election_win(rec, em)) {
2112                 if (!rec->send_election_te) {
2113                         rec->send_election_te = event_add_timed(ctdb->ev, rec,
2114                                                                 timeval_current_ofs(0, 500000),
2115                                                                 election_send_request, rec);
2116                 }
2117                 talloc_free(mem_ctx);
2118                 /*unban_all_nodes(ctdb);*/
2119                 return;
2120         }
2121
2122         /* we didn't win */
2123         talloc_free(rec->send_election_te);
2124         rec->send_election_te = NULL;
2125
2126         if (ctdb->tunable.verify_recovery_lock != 0) {
2127                 /* release the recmaster lock */
2128                 if (em->pnn != ctdb->pnn &&
2129                     ctdb->recovery_lock_fd != -1) {
2130                         close(ctdb->recovery_lock_fd);
2131                         ctdb->recovery_lock_fd = -1;
2132                         unban_all_nodes(ctdb);
2133                 }
2134         }
2135
2136         /* ok, let that guy become recmaster then */
2137         ret = ctdb_ctrl_setrecmaster(ctdb, CONTROL_TIMEOUT(), ctdb_get_pnn(ctdb), em->pnn);
2138         if (ret != 0) {
2139                 DEBUG(DEBUG_ERR, (__location__ " failed to send recmaster election request"));
2140                 talloc_free(mem_ctx);
2141                 return;
2142         }
2143
2144         talloc_free(mem_ctx);
2145         return;
2146 }
2147
2148
2149 /*
2150   force the start of the election process
2151  */
2152 static void force_election(struct ctdb_recoverd *rec, uint32_t pnn,
2153                            struct ctdb_node_map *nodemap)
2154 {
2155         int ret;
2156         struct ctdb_context *ctdb = rec->ctdb;
2157
2158         DEBUG(DEBUG_INFO,(__location__ " Force an election\n"));
2159
2160         /* set all nodes to recovery mode to stop all internode traffic */
2161         ret = set_recovery_mode(ctdb, rec, nodemap, CTDB_RECOVERY_ACTIVE);
2162         if (ret != 0) {
2163                 DEBUG(DEBUG_ERR, (__location__ " Unable to set recovery mode to active on cluster\n"));
2164                 return;
2165         }
2166
2167         talloc_free(rec->election_timeout);
2168         rec->election_timeout = event_add_timed(ctdb->ev, ctdb,
2169                                                 fast_start ?
2170                                                 timeval_current_ofs(0, 500000) :
2171                                                 timeval_current_ofs(ctdb->tunable.election_timeout, 0),
2172                                                 ctdb_election_timeout, rec);
2173
2174         ret = send_election_request(rec, pnn, true);
2175         if (ret!=0) {
2176                 DEBUG(DEBUG_ERR, (__location__ " failed to initiate recmaster election"));
2177                 return;
2178         }
2179
2180         /* wait for a few seconds to collect all responses */
2181         ctdb_wait_election(rec);
2182 }
2183
2184
2185
2186 /*
2187   handler for when a node changes its flags
2188 */
2189 static void monitor_handler(struct ctdb_context *ctdb, uint64_t srvid,
2190                             TDB_DATA data, void *private_data)
2191 {
2192         int ret;
2193         struct ctdb_node_flag_change *c = (struct ctdb_node_flag_change *)data.dptr;
2194         struct ctdb_node_map *nodemap=NULL;
2195         TALLOC_CTX *tmp_ctx;
2196         uint32_t changed_flags;
2197         int i;
2198         struct ctdb_recoverd *rec = talloc_get_type(private_data, struct ctdb_recoverd);
2199         int disabled_flag_changed;
2200
2201         if (data.dsize != sizeof(*c)) {
2202                 DEBUG(DEBUG_ERR,(__location__ "Invalid data in ctdb_node_flag_change\n"));
2203                 return;
2204         }
2205
2206         tmp_ctx = talloc_new(ctdb);
2207         CTDB_NO_MEMORY_VOID(ctdb, tmp_ctx);
2208
2209         ret = ctdb_ctrl_getnodemap(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, tmp_ctx, &nodemap);
2210         if (ret != 0) {
2211                 DEBUG(DEBUG_ERR,(__location__ "ctdb_ctrl_getnodemap failed in monitor_handler\n"));
2212                 talloc_free(tmp_ctx);
2213                 return;
2214         }
2215
2216
2217         for (i=0;i<nodemap->num;i++) {
2218                 if (nodemap->nodes[i].pnn == c->pnn) break;
2219         }
2220
2221         if (i == nodemap->num) {
2222                 DEBUG(DEBUG_CRIT,(__location__ "Flag change for non-existant node %u\n", c->pnn));
2223                 talloc_free(tmp_ctx);
2224                 return;
2225         }
2226
2227         changed_flags = c->old_flags ^ c->new_flags;
2228
2229         if (nodemap->nodes[i].flags != c->new_flags) {
2230                 DEBUG(DEBUG_NOTICE,("Node %u has changed flags - now 0x%x  was 0x%x\n", c->pnn, c->new_flags, c->old_flags));
2231         }
2232
2233         disabled_flag_changed =  (nodemap->nodes[i].flags ^ c->new_flags) & NODE_FLAGS_DISABLED;
2234
2235         nodemap->nodes[i].flags = c->new_flags;
2236
2237         ret = ctdb_ctrl_getrecmaster(ctdb, tmp_ctx, CONTROL_TIMEOUT(),
2238                                      CTDB_CURRENT_NODE, &ctdb->recovery_master);
2239
2240         if (ret == 0) {
2241                 ret = ctdb_ctrl_getrecmode(ctdb, tmp_ctx, CONTROL_TIMEOUT(),
2242                                            CTDB_CURRENT_NODE, &ctdb->recovery_mode);
2243         }
2244
2245         if (ret == 0 &&
2246             ctdb->recovery_master == ctdb->pnn &&
2247             ctdb->recovery_mode == CTDB_RECOVERY_NORMAL) {
2248                 /* Only do the takeover run if the perm disabled or unhealthy
2249                    flags changed since these will cause an ip failover but not
2250                    a recovery.
2251                    If the node became disconnected or banned this will also
2252                    lead to an ip address failover but that is handled
2253                    during recovery
2254                 */
2255                 if (disabled_flag_changed) {
2256                         rec->need_takeover_run = true;
2257                 }
2258         }
2259
2260         talloc_free(tmp_ctx);
2261 }
2262
2263 /*
2264   handler for when we need to push out flag changes ot all other nodes
2265 */
2266 static void push_flags_handler(struct ctdb_context *ctdb, uint64_t srvid,
2267                             TDB_DATA data, void *private_data)
2268 {
2269         int ret;
2270         struct ctdb_node_flag_change *c = (struct ctdb_node_flag_change *)data.dptr;
2271         struct ctdb_node_map *nodemap=NULL;
2272         TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
2273         uint32_t recmaster;
2274         uint32_t *nodes;
2275
2276         /* find the recovery master */
2277         ret = ctdb_ctrl_getrecmaster(ctdb, tmp_ctx, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, &recmaster);
2278         if (ret != 0) {
2279                 DEBUG(DEBUG_ERR, (__location__ " Unable to get recmaster from local node\n"));
2280                 talloc_free(tmp_ctx);
2281                 return;
2282         }
2283
2284         /* read the node flags from the recmaster */
2285         ret = ctdb_ctrl_getnodemap(ctdb, CONTROL_TIMEOUT(), recmaster, tmp_ctx, &nodemap);
2286         if (ret != 0) {
2287                 DEBUG(DEBUG_ERR, (__location__ " Unable to get nodemap from node %u\n", c->pnn));
2288                 talloc_free(tmp_ctx);
2289                 return;
2290         }
2291         if (c->pnn >= nodemap->num) {
2292                 DEBUG(DEBUG_ERR,(__location__ " Nodemap from recmaster does not contain node %d\n", c->pnn));
2293                 talloc_free(tmp_ctx);
2294                 return;
2295         }
2296
2297         /* send the flags update to all connected nodes */
2298         nodes = list_of_connected_nodes(ctdb, nodemap, tmp_ctx, true);
2299
2300         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_MODIFY_FLAGS,
2301                                       nodes, 0, CONTROL_TIMEOUT(),
2302                                       false, data,
2303                                       NULL, NULL,
2304                                       NULL) != 0) {
2305                 DEBUG(DEBUG_ERR, (__location__ " ctdb_control to modify node flags failed\n"));
2306
2307                 talloc_free(tmp_ctx);
2308                 return;
2309         }
2310
2311         talloc_free(tmp_ctx);
2312 }
2313
2314
2315 struct verify_recmode_normal_data {
2316         uint32_t count;
2317         enum monitor_result status;
2318 };
2319
2320 static void verify_recmode_normal_callback(struct ctdb_client_control_state *state)
2321 {
2322         struct verify_recmode_normal_data *rmdata = talloc_get_type(state->async.private_data, struct verify_recmode_normal_data);
2323
2324
2325         /* one more node has responded with recmode data*/
2326         rmdata->count--;
2327
2328         /* if we failed to get the recmode, then return an error and let
2329            the main loop try again.
2330         */
2331         if (state->state != CTDB_CONTROL_DONE) {
2332                 if (rmdata->status == MONITOR_OK) {
2333                         rmdata->status = MONITOR_FAILED;
2334                 }
2335                 return;
2336         }
2337
2338         /* if we got a response, then the recmode will be stored in the
2339            status field
2340         */
2341         if (state->status != CTDB_RECOVERY_NORMAL) {
2342                 DEBUG(DEBUG_NOTICE, (__location__ " Node:%u was in recovery mode. Restart recovery process\n", state->c->hdr.destnode));
2343                 rmdata->status = MONITOR_RECOVERY_NEEDED;
2344         }
2345
2346         return;
2347 }
2348
2349
2350 /* verify that all nodes are in normal recovery mode */
2351 static enum monitor_result verify_recmode(struct ctdb_context *ctdb, struct ctdb_node_map *nodemap)
2352 {
2353         struct verify_recmode_normal_data *rmdata;
2354         TALLOC_CTX *mem_ctx = talloc_new(ctdb);
2355         struct ctdb_client_control_state *state;
2356         enum monitor_result status;
2357         int j;
2358
2359         rmdata = talloc(mem_ctx, struct verify_recmode_normal_data);
2360         CTDB_NO_MEMORY_FATAL(ctdb, rmdata);
2361         rmdata->count  = 0;
2362         rmdata->status = MONITOR_OK;
2363
2364         /* loop over all active nodes and send an async getrecmode call to
2365            them*/
2366         for (j=0; j<nodemap->num; j++) {
2367                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
2368                         continue;
2369                 }
2370                 state = ctdb_ctrl_getrecmode_send(ctdb, mem_ctx,
2371                                         CONTROL_TIMEOUT(),
2372                                         nodemap->nodes[j].pnn);
2373                 if (state == NULL) {
2374                         /* we failed to send the control, treat this as
2375                            an error and try again next iteration
2376                         */
2377                         DEBUG(DEBUG_ERR,("Failed to call ctdb_ctrl_getrecmode_send during monitoring\n"));
2378                         talloc_free(mem_ctx);
2379                         return MONITOR_FAILED;
2380                 }
2381
2382                 /* set up the callback functions */
2383                 state->async.fn = verify_recmode_normal_callback;
2384                 state->async.private_data = rmdata;
2385
2386                 /* one more control to wait for to complete */
2387                 rmdata->count++;
2388         }
2389
2390
2391         /* now wait for up to the maximum number of seconds allowed
2392            or until all nodes we expect a response from has replied
2393         */
2394         while (rmdata->count > 0) {
2395                 event_loop_once(ctdb->ev);
2396         }
2397
2398         status = rmdata->status;
2399         talloc_free(mem_ctx);
2400         return status;
2401 }
2402
2403
2404 struct verify_recmaster_data {
2405         struct ctdb_recoverd *rec;
2406         uint32_t count;
2407         uint32_t pnn;
2408         enum monitor_result status;
2409 };
2410
2411 static void verify_recmaster_callback(struct ctdb_client_control_state *state)
2412 {
2413         struct verify_recmaster_data *rmdata = talloc_get_type(state->async.private_data, struct verify_recmaster_data);
2414
2415
2416         /* one more node has responded with recmaster data*/
2417         rmdata->count--;
2418
2419         /* if we failed to get the recmaster, then return an error and let
2420            the main loop try again.
2421         */
2422         if (state->state != CTDB_CONTROL_DONE) {
2423                 if (rmdata->status == MONITOR_OK) {
2424                         rmdata->status = MONITOR_FAILED;
2425                 }
2426                 return;
2427         }
2428
2429         /* if we got a response, then the recmaster will be stored in the
2430            status field
2431         */
2432         if (state->status != rmdata->pnn) {
2433                 DEBUG(DEBUG_ERR,("Node %d does not agree we are the recmaster. Need a new recmaster election\n", state->c->hdr.destnode));
2434                 ctdb_set_culprit(rmdata->rec, state->c->hdr.destnode);
2435                 rmdata->status = MONITOR_ELECTION_NEEDED;
2436         }
2437
2438         return;
2439 }
2440
2441
2442 /* verify that all nodes agree that we are the recmaster */
2443 static enum monitor_result verify_recmaster(struct ctdb_recoverd *rec, struct ctdb_node_map *nodemap, uint32_t pnn)
2444 {
2445         struct ctdb_context *ctdb = rec->ctdb;
2446         struct verify_recmaster_data *rmdata;
2447         TALLOC_CTX *mem_ctx = talloc_new(ctdb);
2448         struct ctdb_client_control_state *state;
2449         enum monitor_result status;
2450         int j;
2451
2452         rmdata = talloc(mem_ctx, struct verify_recmaster_data);
2453         CTDB_NO_MEMORY_FATAL(ctdb, rmdata);
2454         rmdata->rec    = rec;
2455         rmdata->count  = 0;
2456         rmdata->pnn    = pnn;
2457         rmdata->status = MONITOR_OK;
2458
2459         /* loop over all active nodes and send an async getrecmaster call to
2460            them*/
2461         for (j=0; j<nodemap->num; j++) {
2462                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
2463                         continue;
2464                 }
2465                 state = ctdb_ctrl_getrecmaster_send(ctdb, mem_ctx,
2466                                         CONTROL_TIMEOUT(),
2467                                         nodemap->nodes[j].pnn);
2468                 if (state == NULL) {
2469                         /* we failed to send the control, treat this as
2470                            an error and try again next iteration
2471                         */
2472                         DEBUG(DEBUG_ERR,("Failed to call ctdb_ctrl_getrecmaster_send during monitoring\n"));
2473                         talloc_free(mem_ctx);
2474                         return MONITOR_FAILED;
2475                 }
2476
2477                 /* set up the callback functions */
2478                 state->async.fn = verify_recmaster_callback;
2479                 state->async.private_data = rmdata;
2480
2481                 /* one more control to wait for to complete */
2482                 rmdata->count++;
2483         }
2484
2485
2486         /* now wait for up to the maximum number of seconds allowed
2487            or until all nodes we expect a response from has replied
2488         */
2489         while (rmdata->count > 0) {
2490                 event_loop_once(ctdb->ev);
2491         }
2492
2493         status = rmdata->status;
2494         talloc_free(mem_ctx);
2495         return status;
2496 }
2497
2498
2499 /* called to check that the local allocation of public ip addresses is ok.
2500 */
2501 static int verify_local_ip_allocation(struct ctdb_context *ctdb, struct ctdb_recoverd *rec, uint32_t pnn, struct ctdb_node_map *nodemap)
2502 {
2503         TALLOC_CTX *mem_ctx = talloc_new(NULL);
2504         struct ctdb_control_get_ifaces *ifaces = NULL;
2505         struct ctdb_all_public_ips *ips = NULL;
2506         struct ctdb_uptime *uptime1 = NULL;
2507         struct ctdb_uptime *uptime2 = NULL;
2508         int ret, j;
2509         bool need_iface_check = false;
2510         bool need_takeover_run = false;
2511
2512         ret = ctdb_ctrl_uptime(ctdb, mem_ctx, CONTROL_TIMEOUT(),
2513                                 CTDB_CURRENT_NODE, &uptime1);
2514         if (ret != 0) {
2515                 DEBUG(DEBUG_ERR, ("Unable to get uptime from local node %u\n", pnn));
2516                 talloc_free(mem_ctx);
2517                 return -1;
2518         }
2519
2520
2521         /* read the interfaces from the local node */
2522         ret = ctdb_ctrl_get_ifaces(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, mem_ctx, &ifaces);
2523         if (ret != 0) {
2524                 DEBUG(DEBUG_ERR, ("Unable to get interfaces from local node %u\n", pnn));
2525                 talloc_free(mem_ctx);
2526                 return -1;
2527         }
2528
2529         if (!rec->ifaces) {
2530                 need_iface_check = true;
2531         } else if (rec->ifaces->num != ifaces->num) {
2532                 need_iface_check = true;
2533         } else if (memcmp(rec->ifaces, ifaces, talloc_get_size(ifaces)) != 0) {
2534                 need_iface_check = true;
2535         }
2536
2537         if (need_iface_check) {
2538                 DEBUG(DEBUG_NOTICE, ("The interfaces status has changed on "
2539                                      "local node %u - force takeover run\n",
2540                                      pnn));
2541                 need_takeover_run = true;
2542         }
2543
2544         /* read the ip allocation from the local node */
2545         ret = ctdb_ctrl_get_public_ips(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, mem_ctx, &ips);
2546         if (ret != 0) {
2547                 DEBUG(DEBUG_ERR, ("Unable to get public ips from local node %u\n", pnn));
2548                 talloc_free(mem_ctx);
2549                 return -1;
2550         }
2551
2552         ret = ctdb_ctrl_uptime(ctdb, mem_ctx, CONTROL_TIMEOUT(),
2553                                 CTDB_CURRENT_NODE, &uptime2);
2554         if (ret != 0) {
2555                 DEBUG(DEBUG_ERR, ("Unable to get uptime from local node %u\n", pnn));
2556                 talloc_free(mem_ctx);
2557                 return -1;
2558         }
2559
2560         /* skip the check if the startrecovery time has changed */
2561         if (timeval_compare(&uptime1->last_recovery_started,
2562                             &uptime2->last_recovery_started) != 0) {
2563                 DEBUG(DEBUG_NOTICE, (__location__ " last recovery time changed while we read the public ip list. skipping public ip address check\n"));
2564                 talloc_free(mem_ctx);
2565                 return 0;
2566         }
2567
2568         /* skip the check if the endrecovery time has changed */
2569         if (timeval_compare(&uptime1->last_recovery_finished,
2570                             &uptime2->last_recovery_finished) != 0) {
2571                 DEBUG(DEBUG_NOTICE, (__location__ " last recovery time changed while we read the public ip list. skipping public ip address check\n"));
2572                 talloc_free(mem_ctx);
2573                 return 0;
2574         }
2575
2576         /* skip the check if we have started but not finished recovery */
2577         if (timeval_compare(&uptime1->last_recovery_finished,
2578                             &uptime1->last_recovery_started) != 1) {
2579                 DEBUG(DEBUG_INFO, (__location__ " in the middle of recovery or ip reallocation. skipping public ip address check\n"));
2580                 talloc_free(mem_ctx);
2581
2582                 return 0;
2583         }
2584
2585         talloc_free(rec->ifaces);
2586         rec->ifaces = talloc_steal(rec, ifaces);
2587
2588         /* verify that we have the ip addresses we should have
2589            and we dont have ones we shouldnt have.
2590            if we find an inconsistency we set recmode to
2591            active on the local node and wait for the recmaster
2592            to do a full blown recovery.
2593            also if the pnn is -1 and we are healthy and can host the ip
2594            we also request a ip reallocation.
2595         */
2596         if (ctdb->tunable.disable_ip_failover == 0) {
2597                 for (j=0; j<ips->num; j++) {
2598                         if (ips->ips[j].pnn == -1 && nodemap->nodes[pnn].flags == 0) {
2599                                 DEBUG(DEBUG_CRIT,("Public address '%s' is not assigned and we could serve this ip\n",
2600                                                 ctdb_addr_to_str(&ips->ips[j].addr)));
2601                                 need_takeover_run = true;
2602                         } else if (ips->ips[j].pnn == pnn) {
2603                                 if (!ctdb_sys_have_ip(&ips->ips[j].addr)) {
2604                                         DEBUG(DEBUG_CRIT,("Public address '%s' is missing and we should serve this ip\n",
2605                                                 ctdb_addr_to_str(&ips->ips[j].addr)));
2606                                         need_takeover_run = true;
2607                                 }
2608                         } else {
2609                                 if (ctdb_sys_have_ip(&ips->ips[j].addr)) {
2610                                         DEBUG(DEBUG_CRIT,("We are still serving a public address '%s' that we should not be serving.\n",
2611                                                 ctdb_addr_to_str(&ips->ips[j].addr)));
2612                                         need_takeover_run = true;
2613                                 }
2614                         }
2615                 }
2616         }
2617
2618         if (need_takeover_run) {
2619                 struct takeover_run_reply rd;
2620                 TDB_DATA data;
2621
2622                 DEBUG(DEBUG_CRIT,("Trigger takeoverrun\n"));
2623
2624                 rd.pnn = ctdb->pnn;
2625                 rd.srvid = 0;
2626                 data.dptr = (uint8_t *)&rd;
2627                 data.dsize = sizeof(rd);
2628
2629                 ret = ctdb_client_send_message(ctdb, rec->recmaster, CTDB_SRVID_TAKEOVER_RUN, data);
2630                 if (ret != 0) {
2631                         DEBUG(DEBUG_ERR,(__location__ " Failed to send ipreallocate to recmaster :%d\n", (int)rec->recmaster));
2632                 }
2633         }
2634         talloc_free(mem_ctx);
2635         return 0;
2636 }
2637
2638
2639 static void async_getnodemap_callback(struct ctdb_context *ctdb, uint32_t node_pnn, int32_t res, TDB_DATA outdata, void *callback_data)
2640 {
2641         struct ctdb_node_map **remote_nodemaps = callback_data;
2642
2643         if (node_pnn >= ctdb->num_nodes) {
2644                 DEBUG(DEBUG_ERR,(__location__ " pnn from invalid node\n"));
2645                 return;
2646         }
2647
2648         remote_nodemaps[node_pnn] = (struct ctdb_node_map *)talloc_steal(remote_nodemaps, outdata.dptr);
2649
2650 }
2651
2652 static int get_remote_nodemaps(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
2653         struct ctdb_node_map *nodemap,
2654         struct ctdb_node_map **remote_nodemaps)
2655 {
2656         uint32_t *nodes;
2657
2658         nodes = list_of_active_nodes(ctdb, nodemap, mem_ctx, true);
2659         if (ctdb_client_async_control(ctdb, CTDB_CONTROL_GET_NODEMAP,
2660                                         nodes, 0,
2661                                         CONTROL_TIMEOUT(), false, tdb_null,
2662                                         async_getnodemap_callback,
2663                                         NULL,
2664                                         remote_nodemaps) != 0) {
2665                 DEBUG(DEBUG_ERR, (__location__ " Unable to pull all remote nodemaps\n"));
2666
2667                 return -1;
2668         }
2669
2670         return 0;
2671 }
2672
2673 enum reclock_child_status { RECLOCK_CHECKING, RECLOCK_OK, RECLOCK_FAILED, RECLOCK_TIMEOUT};
2674 struct ctdb_check_reclock_state {
2675         struct ctdb_context *ctdb;
2676         struct timeval start_time;
2677         int fd[2];
2678         pid_t child;
2679         struct timed_event *te;
2680         struct fd_event *fde;
2681         enum reclock_child_status status;
2682 };
2683
2684 /* when we free the reclock state we must kill any child process.
2685 */
2686 static int check_reclock_destructor(struct ctdb_check_reclock_state *state)
2687 {
2688         struct ctdb_context *ctdb = state->ctdb;
2689
2690         ctdb_ctrl_report_recd_lock_latency(ctdb, CONTROL_TIMEOUT(), timeval_elapsed(&state->start_time));
2691
2692         if (state->fd[0] != -1) {
2693                 close(state->fd[0]);
2694                 state->fd[0] = -1;
2695         }
2696         if (state->fd[1] != -1) {
2697                 close(state->fd[1]);
2698                 state->fd[1] = -1;
2699         }
2700         kill(state->child, SIGKILL);
2701         return 0;
2702 }
2703
2704 /*
2705   called if our check_reclock child times out. this would happen if
2706   i/o to the reclock file blocks.
2707  */
2708 static void ctdb_check_reclock_timeout(struct event_context *ev, struct timed_event *te,
2709                                          struct timeval t, void *private_data)
2710 {
2711         struct ctdb_check_reclock_state *state = talloc_get_type(private_data,
2712                                            struct ctdb_check_reclock_state);
2713
2714         DEBUG(DEBUG_ERR,(__location__ " check_reclock child process hung/timedout CFS slow to grant locks?\n"));
2715         state->status = RECLOCK_TIMEOUT;
2716 }
2717
2718 /* this is called when the child process has completed checking the reclock
2719    file and has written data back to us through the pipe.
2720 */
2721 static void reclock_child_handler(struct event_context *ev, struct fd_event *fde,
2722                              uint16_t flags, void *private_data)
2723 {
2724         struct ctdb_check_reclock_state *state= talloc_get_type(private_data,
2725                                              struct ctdb_check_reclock_state);
2726         char c = 0;
2727         int ret;
2728
2729         /* we got a response from our child process so we can abort the
2730            timeout.
2731         */
2732         talloc_free(state->te);
2733         state->te = NULL;
2734
2735         ret = read(state->fd[0], &c, 1);
2736         if (ret != 1 || c != RECLOCK_OK) {
2737                 DEBUG(DEBUG_ERR,(__location__ " reclock child process returned error %d\n", c));
2738                 state->status = RECLOCK_FAILED;
2739
2740                 return;
2741         }
2742
2743         state->status = RECLOCK_OK;
2744         return;
2745 }
2746
2747 static int check_recovery_lock(struct ctdb_context *ctdb)
2748 {
2749         int ret;
2750         struct ctdb_check_reclock_state *state;
2751         pid_t parent = getpid();
2752
2753         if (ctdb->recovery_lock_fd == -1) {
2754                 DEBUG(DEBUG_CRIT,("recovery master doesn't have the recovery lock\n"));
2755                 return -1;
2756         }
2757
2758         state = talloc(ctdb, struct ctdb_check_reclock_state);
2759         CTDB_NO_MEMORY(ctdb, state);
2760
2761         state->ctdb = ctdb;
2762         state->start_time = timeval_current();
2763         state->status = RECLOCK_CHECKING;
2764         state->fd[0] = -1;
2765         state->fd[1] = -1;
2766
2767         ret = pipe(state->fd);
2768         if (ret != 0) {
2769                 talloc_free(state);
2770                 DEBUG(DEBUG_CRIT,(__location__ " Failed to open pipe for check_reclock child\n"));
2771                 return -1;
2772         }
2773
2774         state->child = ctdb_fork(ctdb);
2775         if (state->child == (pid_t)-1) {
2776                 DEBUG(DEBUG_CRIT,(__location__ " fork() failed in check_reclock child\n"));
2777                 close(state->fd[0]);
2778                 state->fd[0] = -1;
2779                 close(state->fd[1]);
2780                 state->fd[1] = -1;
2781                 talloc_free(state);
2782                 return -1;
2783         }
2784
2785         if (state->child == 0) {
2786                 char cc = RECLOCK_OK;
2787                 close(state->fd[0]);
2788                 state->fd[0] = -1;
2789
2790                 debug_extra = talloc_asprintf(NULL, "recovery-lock:");
2791                 if (pread(ctdb->recovery_lock_fd, &cc, 1, 0) == -1) {
2792                         DEBUG(DEBUG_CRIT,("failed read from recovery_lock_fd - %s\n", strerror(errno)));
2793                         cc = RECLOCK_FAILED;
2794                 }
2795
2796                 write(state->fd[1], &cc, 1);
2797                 /* make sure we die when our parent dies */
2798                 while (kill(parent, 0) == 0 || errno != ESRCH) {
2799                         sleep(5);
2800                         write(state->fd[1], &cc, 1);
2801                 }
2802                 _exit(0);
2803         }
2804         close(state->fd[1]);
2805         state->fd[1] = -1;
2806         set_close_on_exec(state->fd[0]);
2807
2808         DEBUG(DEBUG_DEBUG, (__location__ " Created PIPE FD:%d for check_recovery_lock\n", state->fd[0]));
2809
2810         talloc_set_destructor(state, check_reclock_destructor);
2811
2812         state->te = event_add_timed(ctdb->ev, state, timeval_current_ofs(15, 0),
2813                                     ctdb_check_reclock_timeout, state);
2814         if (state->te == NULL) {
2815                 DEBUG(DEBUG_CRIT,(__location__ " Failed to create a timed event for reclock child\n"));
2816                 talloc_free(state);
2817                 return -1;
2818         }
2819
2820         state->fde = event_add_fd(ctdb->ev, state, state->fd[0],
2821                                 EVENT_FD_READ,
2822                                 reclock_child_handler,
2823                                 (void *)state);
2824
2825         if (state->fde == NULL) {
2826                 DEBUG(DEBUG_CRIT,(__location__ " Failed to create an fd event for reclock child\n"));
2827                 talloc_free(state);
2828                 return -1;
2829         }
2830         tevent_fd_set_auto_close(state->fde);
2831
2832         while (state->status == RECLOCK_CHECKING) {
2833                 event_loop_once(ctdb->ev);
2834         }
2835
2836         if (state->status == RECLOCK_FAILED) {
2837                 DEBUG(DEBUG_ERR,(__location__ " reclock child failed when checking file\n"));
2838                 close(ctdb->recovery_lock_fd);
2839                 ctdb->recovery_lock_fd = -1;
2840                 talloc_free(state);
2841                 return -1;
2842         }
2843
2844         talloc_free(state);
2845         return 0;
2846 }
2847
2848 static int update_recovery_lock_file(struct ctdb_context *ctdb)
2849 {
2850         TALLOC_CTX *tmp_ctx = talloc_new(NULL);
2851         const char *reclockfile;
2852
2853         if (ctdb_ctrl_getreclock(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, tmp_ctx, &reclockfile) != 0) {
2854                 DEBUG(DEBUG_ERR,("Failed to read reclock file from daemon\n"));
2855                 talloc_free(tmp_ctx);
2856                 return -1;
2857         }
2858
2859         if (reclockfile == NULL) {
2860                 if (ctdb->recovery_lock_file != NULL) {
2861                         DEBUG(DEBUG_ERR,("Reclock file disabled\n"));
2862                         talloc_free(ctdb->recovery_lock_file);
2863                         ctdb->recovery_lock_file = NULL;
2864                         if (ctdb->recovery_lock_fd != -1) {
2865                                 close(ctdb->recovery_lock_fd);
2866                                 ctdb->recovery_lock_fd = -1;
2867                         }
2868                 }
2869                 ctdb->tunable.verify_recovery_lock = 0;
2870                 talloc_free(tmp_ctx);
2871                 return 0;
2872         }
2873
2874         if (ctdb->recovery_lock_file == NULL) {
2875                 ctdb->recovery_lock_file = talloc_strdup(ctdb, reclockfile);
2876                 if (ctdb->recovery_lock_fd != -1) {
2877                         close(ctdb->recovery_lock_fd);
2878                         ctdb->recovery_lock_fd = -1;
2879                 }
2880                 talloc_free(tmp_ctx);
2881                 return 0;
2882         }
2883
2884
2885         if (!strcmp(reclockfile, ctdb->recovery_lock_file)) {
2886                 talloc_free(tmp_ctx);
2887                 return 0;
2888         }
2889
2890         talloc_free(ctdb->recovery_lock_file);
2891         ctdb->recovery_lock_file = talloc_strdup(ctdb, reclockfile);
2892         ctdb->tunable.verify_recovery_lock = 0;
2893         if (ctdb->recovery_lock_fd != -1) {
2894                 close(ctdb->recovery_lock_fd);
2895                 ctdb->recovery_lock_fd = -1;
2896         }
2897
2898         talloc_free(tmp_ctx);
2899         return 0;
2900 }
2901
2902 static void main_loop(struct ctdb_context *ctdb, struct ctdb_recoverd *rec,
2903                       TALLOC_CTX *mem_ctx)
2904 {
2905         uint32_t pnn;
2906         struct ctdb_node_map *nodemap=NULL;
2907         struct ctdb_node_map *recmaster_nodemap=NULL;
2908         struct ctdb_node_map **remote_nodemaps=NULL;
2909         struct ctdb_vnn_map *vnnmap=NULL;
2910         struct ctdb_vnn_map *remote_vnnmap=NULL;
2911         int32_t debug_level;
2912         int i, j, ret;
2913
2914
2915
2916         /* verify that the main daemon is still running */
2917         if (kill(ctdb->ctdbd_pid, 0) != 0) {
2918                 DEBUG(DEBUG_CRIT,("CTDB daemon is no longer available. Shutting down recovery daemon\n"));
2919                 exit(-1);
2920         }
2921
2922         /* ping the local daemon to tell it we are alive */
2923         ctdb_ctrl_recd_ping(ctdb);
2924
2925         if (rec->election_timeout) {
2926                 /* an election is in progress */
2927                 return;
2928         }
2929
2930         /* read the debug level from the parent and update locally */
2931         ret = ctdb_ctrl_get_debuglevel(ctdb, CTDB_CURRENT_NODE, &debug_level);
2932         if (ret !=0) {
2933                 DEBUG(DEBUG_ERR, (__location__ " Failed to read debuglevel from parent\n"));
2934                 return;
2935         }
2936         LogLevel = debug_level;
2937
2938
2939         /* We must check if we need to ban a node here but we want to do this
2940            as early as possible so we dont wait until we have pulled the node
2941            map from the local node. thats why we have the hardcoded value 20
2942         */
2943         for (i=0; i<ctdb->num_nodes; i++) {
2944                 struct ctdb_banning_state *ban_state;
2945
2946                 if (ctdb->nodes[i]->ban_state == NULL) {
2947                         continue;
2948                 }
2949                 ban_state = (struct ctdb_banning_state *)ctdb->nodes[i]->ban_state;
2950                 if (ban_state->count < 20) {
2951                         continue;
2952                 }
2953                 DEBUG(DEBUG_NOTICE,("Node %u has caused %u recoveries recently - banning it for %u seconds\n",
2954                         ctdb->nodes[i]->pnn, ban_state->count,
2955                         ctdb->tunable.recovery_ban_period));
2956                 ctdb_ban_node(rec, ctdb->nodes[i]->pnn, ctdb->tunable.recovery_ban_period);
2957                 ban_state->count = 0;
2958         }
2959
2960         /* get relevant tunables */
2961         ret = ctdb_ctrl_get_all_tunables(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, &ctdb->tunable);
2962         if (ret != 0) {
2963                 DEBUG(DEBUG_ERR,("Failed to get tunables - retrying\n"));
2964                 return;
2965         }
2966
2967         /* get the current recovery lock file from the server */
2968         if (update_recovery_lock_file(ctdb) != 0) {
2969                 DEBUG(DEBUG_ERR,("Failed to update the recovery lock file\n"));
2970                 return;
2971         }
2972
2973         /* Make sure that if recovery lock verification becomes disabled when
2974            we close the file
2975         */
2976         if (ctdb->tunable.verify_recovery_lock == 0) {
2977                 if (ctdb->recovery_lock_fd != -1) {
2978                         close(ctdb->recovery_lock_fd);
2979                         ctdb->recovery_lock_fd = -1;
2980                 }
2981         }
2982
2983         pnn = ctdb_ctrl_getpnn(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE);
2984         if (pnn == (uint32_t)-1) {
2985                 DEBUG(DEBUG_ERR,("Failed to get local pnn - retrying\n"));
2986                 return;
2987         }
2988
2989         /* get the vnnmap */
2990         ret = ctdb_ctrl_getvnnmap(ctdb, CONTROL_TIMEOUT(), pnn, mem_ctx, &vnnmap);
2991         if (ret != 0) {
2992                 DEBUG(DEBUG_ERR, (__location__ " Unable to get vnnmap from node %u\n", pnn));
2993                 return;
2994         }
2995
2996
2997         /* get number of nodes */
2998         if (rec->nodemap) {
2999                 talloc_free(rec->nodemap);
3000                 rec->nodemap = NULL;
3001                 nodemap=NULL;
3002         }
3003         ret = ctdb_ctrl_getnodemap(ctdb, CONTROL_TIMEOUT(), pnn, rec, &rec->nodemap);
3004         if (ret != 0) {
3005                 DEBUG(DEBUG_ERR, (__location__ " Unable to get nodemap from node %u\n", pnn));
3006                 return;
3007         }
3008         nodemap = rec->nodemap;
3009
3010         /* check which node is the recovery master */
3011         ret = ctdb_ctrl_getrecmaster(ctdb, mem_ctx, CONTROL_TIMEOUT(), pnn, &rec->recmaster);
3012         if (ret != 0) {
3013                 DEBUG(DEBUG_ERR, (__location__ " Unable to get recmaster from node %u\n", pnn));
3014                 return;
3015         }
3016
3017         /* if we are not the recmaster we can safely ignore any ip reallocate requests */
3018         if (rec->recmaster != pnn) {
3019                 if (rec->ip_reallocate_ctx != NULL) {
3020                         talloc_free(rec->ip_reallocate_ctx);
3021                         rec->ip_reallocate_ctx = NULL;
3022                         rec->reallocate_callers = NULL;
3023                 }
3024         }
3025         /* if there are takeovers requested, perform it and notify the waiters */
3026         if (rec->reallocate_callers) {
3027                 process_ipreallocate_requests(ctdb, rec);
3028         }
3029
3030         if (rec->recmaster == (uint32_t)-1) {
3031                 DEBUG(DEBUG_NOTICE,(__location__ " Initial recovery master set - forcing election\n"));
3032                 force_election(rec, pnn, nodemap);
3033                 return;
3034         }
3035
3036
3037         /* if the local daemon is STOPPED, we verify that the databases are
3038            also frozen and thet the recmode is set to active
3039         */
3040         if (nodemap->nodes[pnn].flags & NODE_FLAGS_STOPPED) {
3041                 ret = ctdb_ctrl_getrecmode(ctdb, mem_ctx, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, &ctdb->recovery_mode);
3042                 if (ret != 0) {
3043                         DEBUG(DEBUG_ERR,(__location__ " Failed to read recmode from local node\n"));
3044                 }
3045                 if (ctdb->recovery_mode == CTDB_RECOVERY_NORMAL) {
3046                         DEBUG(DEBUG_ERR,("Node is stopped but recovery mode is not active. Activate recovery mode and lock databases\n"));
3047
3048                         ret = ctdb_ctrl_freeze_priority(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, 1);
3049                         if (ret != 0) {
3050                                 DEBUG(DEBUG_ERR,(__location__ " Failed to freeze node due to node being STOPPED\n"));
3051                                 return;
3052                         }
3053                         ret = ctdb_ctrl_setrecmode(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE, CTDB_RECOVERY_ACTIVE);
3054                         if (ret != 0) {
3055                                 DEBUG(DEBUG_ERR,(__location__ " Failed to activate recovery mode due to node being stopped\n"));
3056
3057                                 return;
3058                         }
3059                         return;
3060                 }
3061         }
3062         /* If the local node is stopped, verify we are not the recmaster
3063            and yield this role if so
3064         */
3065         if ((nodemap->nodes[pnn].flags & NODE_FLAGS_STOPPED) && (rec->recmaster == pnn)) {
3066                 DEBUG(DEBUG_ERR,("Local node is STOPPED. Yielding recmaster role\n"));
3067                 force_election(rec, pnn, nodemap);
3068                 return;
3069         }
3070
3071         /* check that we (recovery daemon) and the local ctdb daemon
3072            agrees on whether we are banned or not
3073         */
3074 //qqq
3075
3076         /* remember our own node flags */
3077         rec->node_flags = nodemap->nodes[pnn].flags;
3078
3079         /* count how many active nodes there are */
3080         rec->num_active    = 0;
3081         rec->num_connected = 0;
3082         for (i=0; i<nodemap->num; i++) {
3083                 if (!(nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE)) {
3084                         rec->num_active++;
3085                 }
3086                 if (!(nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED)) {
3087                         rec->num_connected++;
3088                 }
3089         }
3090
3091
3092         /* verify that the recmaster node is still active */
3093         for (j=0; j<nodemap->num; j++) {
3094                 if (nodemap->nodes[j].pnn==rec->recmaster) {
3095                         break;
3096                 }
3097         }
3098
3099         if (j == nodemap->num) {
3100                 DEBUG(DEBUG_ERR, ("Recmaster node %u not in list. Force reelection\n", rec->recmaster));
3101                 force_election(rec, pnn, nodemap);
3102                 return;
3103         }
3104
3105         /* if recovery master is disconnected we must elect a new recmaster */
3106         if (nodemap->nodes[j].flags & NODE_FLAGS_DISCONNECTED) {
3107                 DEBUG(DEBUG_NOTICE, ("Recmaster node %u is disconnected. Force reelection\n", nodemap->nodes[j].pnn));
3108                 force_election(rec, pnn, nodemap);
3109                 return;
3110         }
3111
3112         /* grap the nodemap from the recovery master to check if it is banned */
3113         ret = ctdb_ctrl_getnodemap(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn,
3114                                    mem_ctx, &recmaster_nodemap);
3115         if (ret != 0) {
3116                 DEBUG(DEBUG_ERR, (__location__ " Unable to get nodemap from recovery master %u\n",
3117                           nodemap->nodes[j].pnn));
3118                 return;
3119         }
3120
3121
3122         if (recmaster_nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
3123                 DEBUG(DEBUG_NOTICE, ("Recmaster node %u no longer available. Force reelection\n", nodemap->nodes[j].pnn));
3124                 force_election(rec, pnn, nodemap);
3125                 return;
3126         }
3127
3128
3129         /* verify that we have all ip addresses we should have and we dont
3130          * have addresses we shouldnt have.
3131          */
3132         if (ctdb->tunable.disable_ip_failover == 0) {
3133                 if (rec->ip_check_disable_ctx == NULL) {
3134                         if (verify_local_ip_allocation(ctdb, rec, pnn, nodemap) != 0) {
3135                                 DEBUG(DEBUG_ERR, (__location__ " Public IPs were inconsistent.\n"));
3136                         }
3137                 }
3138         }
3139
3140
3141         /* if we are not the recmaster then we do not need to check
3142            if recovery is needed
3143          */
3144         if (pnn != rec->recmaster) {
3145                 return;
3146         }
3147
3148
3149         /* ensure our local copies of flags are right */
3150         ret = update_local_flags(rec, nodemap);
3151         if (ret == MONITOR_ELECTION_NEEDED) {
3152                 DEBUG(DEBUG_NOTICE,("update_local_flags() called for a re-election.\n"));
3153                 force_election(rec, pnn, nodemap);
3154                 return;
3155         }
3156         if (ret != MONITOR_OK) {
3157                 DEBUG(DEBUG_ERR,("Unable to update local flags\n"));
3158                 return;
3159         }
3160
3161         if (ctdb->num_nodes != nodemap->num) {
3162                 DEBUG(DEBUG_ERR, (__location__ " ctdb->num_nodes (%d) != nodemap->num (%d) reloading nodes file\n", ctdb->num_nodes, nodemap->num));
3163                 reload_nodes_file(ctdb);
3164                 return;
3165         }
3166
3167         /* verify that all active nodes agree that we are the recmaster */
3168         switch (verify_recmaster(rec, nodemap, pnn)) {
3169         case MONITOR_RECOVERY_NEEDED:
3170                 /* can not happen */
3171                 return;
3172         case MONITOR_ELECTION_NEEDED:
3173                 force_election(rec, pnn, nodemap);
3174                 return;
3175         case MONITOR_OK:
3176                 break;
3177         case MONITOR_FAILED:
3178                 return;
3179         }
3180
3181
3182         if (rec->need_recovery) {
3183                 /* a previous recovery didn't finish */
3184                 do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3185                 return;
3186         }
3187
3188         /* verify that all active nodes are in normal mode
3189            and not in recovery mode
3190         */
3191         switch (verify_recmode(ctdb, nodemap)) {
3192         case MONITOR_RECOVERY_NEEDED:
3193                 do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3194                 return;
3195         case MONITOR_FAILED:
3196                 return;
3197         case MONITOR_ELECTION_NEEDED:
3198                 /* can not happen */
3199         case MONITOR_OK:
3200                 break;
3201         }
3202
3203
3204         if (ctdb->tunable.verify_recovery_lock != 0) {
3205                 /* we should have the reclock - check its not stale */
3206                 ret = check_recovery_lock(ctdb);
3207                 if (ret != 0) {
3208                         DEBUG(DEBUG_ERR,("Failed check_recovery_lock. Force a recovery\n"));
3209                         ctdb_set_culprit(rec, ctdb->pnn);
3210                         do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3211                         return;
3212                 }
3213         }
3214
3215         /* get the nodemap for all active remote nodes
3216          */
3217         remote_nodemaps = talloc_array(mem_ctx, struct ctdb_node_map *, nodemap->num);
3218         if (remote_nodemaps == NULL) {
3219                 DEBUG(DEBUG_ERR, (__location__ " failed to allocate remote nodemap array\n"));
3220                 return;
3221         }
3222         for(i=0; i<nodemap->num; i++) {
3223                 remote_nodemaps[i] = NULL;
3224         }
3225         if (get_remote_nodemaps(ctdb, mem_ctx, nodemap, remote_nodemaps) != 0) {
3226                 DEBUG(DEBUG_ERR,(__location__ " Failed to read remote nodemaps\n"));
3227                 return;
3228         }
3229
3230         /* verify that all other nodes have the same nodemap as we have
3231         */
3232         for (j=0; j<nodemap->num; j++) {
3233                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
3234                         continue;
3235                 }
3236
3237                 if (remote_nodemaps[j] == NULL) {
3238                         DEBUG(DEBUG_ERR,(__location__ " Did not get a remote nodemap for node %d, restarting monitoring\n", j));
3239                         ctdb_set_culprit(rec, j);
3240
3241                         return;
3242                 }
3243
3244                 /* if the nodes disagree on how many nodes there are
3245                    then this is a good reason to try recovery
3246                  */
3247                 if (remote_nodemaps[j]->num != nodemap->num) {
3248                         DEBUG(DEBUG_ERR, (__location__ " Remote node:%u has different node count. %u vs %u of the local node\n",
3249                                   nodemap->nodes[j].pnn, remote_nodemaps[j]->num, nodemap->num));
3250                         ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3251                         do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3252                         return;
3253                 }
3254
3255                 /* if the nodes disagree on which nodes exist and are
3256                    active, then that is also a good reason to do recovery
3257                  */
3258                 for (i=0;i<nodemap->num;i++) {
3259                         if (remote_nodemaps[j]->nodes[i].pnn != nodemap->nodes[i].pnn) {
3260                                 DEBUG(DEBUG_ERR, (__location__ " Remote node:%u has different nodemap pnn for %d (%u vs %u).\n",
3261                                           nodemap->nodes[j].pnn, i,
3262                                           remote_nodemaps[j]->nodes[i].pnn, nodemap->nodes[i].pnn));
3263                                 ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3264                                 do_recovery(rec, mem_ctx, pnn, nodemap,
3265                                             vnnmap);
3266                                 return;
3267                         }
3268                 }
3269
3270                 /* verify the flags are consistent
3271                 */
3272                 for (i=0; i<nodemap->num; i++) {
3273                         if (nodemap->nodes[i].flags & NODE_FLAGS_DISCONNECTED) {
3274                                 continue;
3275                         }
3276
3277                         if (nodemap->nodes[i].flags != remote_nodemaps[j]->nodes[i].flags) {
3278                                 DEBUG(DEBUG_ERR, (__location__ " Remote node:%u has different flags for node %u. It has 0x%02x vs our 0x%02x\n",
3279                                   nodemap->nodes[j].pnn,
3280                                   nodemap->nodes[i].pnn,
3281                                   remote_nodemaps[j]->nodes[i].flags,
3282                                   nodemap->nodes[j].flags));
3283                                 if (i == j) {
3284                                         DEBUG(DEBUG_ERR,("Use flags 0x%02x from remote node %d for cluster update of its own flags\n", remote_nodemaps[j]->nodes[i].flags, j));
3285                                         update_flags_on_all_nodes(ctdb, nodemap, nodemap->nodes[i].pnn, remote_nodemaps[j]->nodes[i].flags);
3286                                         ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3287                                         do_recovery(rec, mem_ctx, pnn, nodemap,
3288                                                     vnnmap);
3289                                         return;
3290                                 } else {
3291                                         DEBUG(DEBUG_ERR,("Use flags 0x%02x from local recmaster node for cluster update of node %d flags\n", nodemap->nodes[i].flags, i));
3292                                         update_flags_on_all_nodes(ctdb, nodemap, nodemap->nodes[i].pnn, nodemap->nodes[i].flags);
3293                                         ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3294                                         do_recovery(rec, mem_ctx, pnn, nodemap,
3295                                                     vnnmap);
3296                                         return;
3297                                 }
3298                         }
3299                 }
3300         }
3301
3302
3303         /* there better be the same number of lmasters in the vnn map
3304            as there are active nodes or we will have to do a recovery
3305          */
3306         if (vnnmap->size != rec->num_active) {
3307                 DEBUG(DEBUG_ERR, (__location__ " The vnnmap count is different from the number of active nodes. %u vs %u\n",
3308                           vnnmap->size, rec->num_active));
3309                 ctdb_set_culprit(rec, ctdb->pnn);
3310                 do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3311                 return;
3312         }
3313
3314         /* verify that all active nodes in the nodemap also exist in
3315            the vnnmap.
3316          */
3317         for (j=0; j<nodemap->num; j++) {
3318                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
3319                         continue;
3320                 }
3321                 if (nodemap->nodes[j].pnn == pnn) {
3322                         continue;
3323                 }
3324
3325                 for (i=0; i<vnnmap->size; i++) {
3326                         if (vnnmap->map[i] == nodemap->nodes[j].pnn) {
3327                                 break;
3328                         }
3329                 }
3330                 if (i == vnnmap->size) {
3331                         DEBUG(DEBUG_ERR, (__location__ " Node %u is active in the nodemap but did not exist in the vnnmap\n",
3332                                   nodemap->nodes[j].pnn));
3333                         ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3334                         do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3335                         return;
3336                 }
3337         }
3338
3339
3340         /* verify that all other nodes have the same vnnmap
3341            and are from the same generation
3342          */
3343         for (j=0; j<nodemap->num; j++) {
3344                 if (nodemap->nodes[j].flags & NODE_FLAGS_INACTIVE) {
3345                         continue;
3346                 }
3347                 if (nodemap->nodes[j].pnn == pnn) {
3348                         continue;
3349                 }
3350
3351                 ret = ctdb_ctrl_getvnnmap(ctdb, CONTROL_TIMEOUT(), nodemap->nodes[j].pnn,
3352                                           mem_ctx, &remote_vnnmap);
3353                 if (ret != 0) {
3354                         DEBUG(DEBUG_ERR, (__location__ " Unable to get vnnmap from remote node %u\n",
3355                                   nodemap->nodes[j].pnn));
3356                         return;
3357                 }
3358
3359                 /* verify the vnnmap generation is the same */
3360                 if (vnnmap->generation != remote_vnnmap->generation) {
3361                         DEBUG(DEBUG_ERR, (__location__ " Remote node %u has different generation of vnnmap. %u vs %u (ours)\n",
3362                                   nodemap->nodes[j].pnn, remote_vnnmap->generation, vnnmap->generation));
3363                         ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3364                         do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3365                         return;
3366                 }
3367
3368                 /* verify the vnnmap size is the same */
3369                 if (vnnmap->size != remote_vnnmap->size) {
3370                         DEBUG(DEBUG_ERR, (__location__ " Remote node %u has different size of vnnmap. %u vs %u (ours)\n",
3371                                   nodemap->nodes[j].pnn, remote_vnnmap->size, vnnmap->size));
3372                         ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3373                         do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3374                         return;
3375                 }
3376
3377                 /* verify the vnnmap is the same */
3378                 for (i=0;i<vnnmap->size;i++) {
3379                         if (remote_vnnmap->map[i] != vnnmap->map[i]) {
3380                                 DEBUG(DEBUG_ERR, (__location__ " Remote node %u has different vnnmap.\n",
3381                                           nodemap->nodes[j].pnn));
3382                                 ctdb_set_culprit(rec, nodemap->nodes[j].pnn);
3383                                 do_recovery(rec, mem_ctx, pnn, nodemap,
3384                                             vnnmap);
3385                                 return;
3386                         }
3387                 }
3388         }
3389
3390         /* we might need to change who has what IP assigned */
3391         if (rec->need_takeover_run) {
3392                 uint32_t culprit = (uint32_t)-1;
3393
3394                 rec->need_takeover_run = false;
3395
3396                 /* update the list of public ips that a node can handle for
3397                    all connected nodes
3398                 */
3399                 ret = ctdb_reload_remote_public_ips(ctdb, rec, nodemap, &culprit);
3400                 if (ret != 0) {
3401                         DEBUG(DEBUG_ERR,("Failed to read public ips from remote node %d\n",
3402                                          culprit));
3403                         rec->need_takeover_run = true;
3404                         return;
3405                 }
3406
3407                 /* execute the "startrecovery" event script on all nodes */
3408                 ret = run_startrecovery_eventscript(rec, nodemap);
3409                 if (ret!=0) {
3410                         DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'startrecovery' event on cluster\n"));
3411                         ctdb_set_culprit(rec, ctdb->pnn);
3412                         do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3413                         return;
3414                 }
3415
3416                 ret = ctdb_takeover_run(ctdb, nodemap);
3417                 if (ret != 0) {
3418                         DEBUG(DEBUG_ERR, (__location__ " Unable to setup public takeover addresses. Try again later\n"));
3419                         return;
3420                 }
3421
3422                 /* execute the "recovered" event script on all nodes */
3423                 ret = run_recovered_eventscript(ctdb, nodemap, "monitor_cluster");
3424 #if 0
3425 // we cant check whether the event completed successfully
3426 // since this script WILL fail if the node is in recovery mode
3427 // and if that race happens, the code here would just cause a second
3428 // cascading recovery.
3429                 if (ret!=0) {
3430                         DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'recovered' event on cluster. Update of public ips failed.\n"));
3431                         ctdb_set_culprit(rec, ctdb->pnn);
3432                         do_recovery(rec, mem_ctx, pnn, nodemap, vnnmap);
3433                 }
3434 #endif
3435         }
3436 }
3437
3438 /*
3439   the main monitoring loop
3440  */
3441 static void monitor_cluster(struct ctdb_context *ctdb)
3442 {
3443         struct ctdb_recoverd *rec;
3444
3445         DEBUG(DEBUG_NOTICE,("monitor_cluster starting\n"));
3446
3447         rec = talloc_zero(ctdb, struct ctdb_recoverd);
3448         CTDB_NO_MEMORY_FATAL(ctdb, rec);
3449
3450         rec->ctdb = ctdb;
3451
3452         rec->priority_time = timeval_current();
3453
3454         /* register a message port for sending memory dumps */
3455         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_MEM_DUMP, mem_dump_handler, rec);
3456
3457         /* register a message port for recovery elections */
3458         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_RECOVERY, election_handler, rec);
3459
3460         /* when nodes are disabled/enabled */
3461         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_SET_NODE_FLAGS, monitor_handler, rec);
3462
3463         /* when we are asked to puch out a flag change */
3464         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_PUSH_NODE_FLAGS, push_flags_handler, rec);
3465
3466         /* register a message port for vacuum fetch */
3467         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_VACUUM_FETCH, vacuum_fetch_handler, rec);
3468
3469         /* register a message port for reloadnodes  */
3470         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_RELOAD_NODES, reload_nodes_handler, rec);
3471
3472         /* register a message port for performing a takeover run */
3473         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_TAKEOVER_RUN, ip_reallocate_handler, rec);
3474
3475         /* register a message port for disabling the ip check for a short while */
3476         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_DISABLE_IP_CHECK, disable_ip_check_handler, rec);
3477
3478         /* register a message port for updating the recovery daemons node assignment for an ip */
3479         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_RECD_UPDATE_IP, recd_update_ip_handler, rec);
3480
3481         for (;;) {
3482                 TALLOC_CTX *mem_ctx = talloc_new(ctdb);
3483                 struct timeval start;
3484                 double elapsed;
3485
3486                 if (!mem_ctx) {
3487                         DEBUG(DEBUG_CRIT,(__location__
3488                                           " Failed to create temp context\n"));
3489                         exit(-1);
3490                 }
3491
3492                 start = timeval_current();
3493                 main_loop(ctdb, rec, mem_ctx);
3494                 talloc_free(mem_ctx);
3495
3496                 /* we only check for recovery once every second */
3497                 elapsed = timeval_elapsed(&start);
3498                 if (elapsed < ctdb->tunable.recover_interval) {
3499                         ctdb_wait_timeout(ctdb, ctdb->tunable.recover_interval
3500                                           - elapsed);
3501                 }
3502         }
3503 }
3504
3505 /*
3506   event handler for when the main ctdbd dies
3507  */
3508 static void ctdb_recoverd_parent(struct event_context *ev, struct fd_event *fde,
3509                                  uint16_t flags, void *private_data)
3510 {
3511         DEBUG(DEBUG_ALERT,("recovery daemon parent died - exiting\n"));
3512         _exit(1);
3513 }
3514
3515 /*
3516   called regularly to verify that the recovery daemon is still running
3517  */
3518 static void ctdb_check_recd(struct event_context *ev, struct timed_event *te,
3519                               struct timeval yt, void *p)
3520 {
3521         struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
3522
3523         if (kill(ctdb->recoverd_pid, 0) != 0) {
3524                 DEBUG(DEBUG_ERR,("Recovery daemon (pid:%d) is no longer running. Trying to restart recovery daemon.\n", (int)ctdb->recoverd_pid));
3525
3526                 event_add_timed(ctdb->ev, ctdb, timeval_zero(),
3527                                 ctdb_restart_recd, ctdb);
3528
3529                 return;
3530         }
3531
3532         event_add_timed(ctdb->ev, ctdb,
3533                         timeval_current_ofs(30, 0),
3534                         ctdb_check_recd, ctdb);
3535 }
3536
3537 static void recd_sig_child_handler(struct event_context *ev,
3538         struct signal_event *se, int signum, int count,
3539         void *dont_care,
3540         void *private_data)
3541 {
3542 //      struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
3543         int status;
3544         pid_t pid = -1;
3545
3546         while (pid != 0) {
3547                 pid = waitpid(-1, &status, WNOHANG);
3548                 if (pid == -1) {
3549                         if (errno != ECHILD) {
3550                                 DEBUG(DEBUG_ERR, (__location__ " waitpid() returned error. errno:%s(%d)\n", strerror(errno),errno));
3551                         }
3552                         return;
3553                 }
3554                 if (pid > 0) {
3555                         DEBUG(DEBUG_DEBUG, ("RECD SIGCHLD from %d\n", (int)pid));
3556                 }
3557         }
3558 }
3559
3560 /*
3561   startup the recovery daemon as a child of the main ctdb daemon
3562  */
3563 int ctdb_start_recoverd(struct ctdb_context *ctdb)
3564 {
3565         int fd[2];
3566         struct signal_event *se;
3567         struct tevent_fd *fde;
3568
3569         if (pipe(fd) != 0) {
3570                 return -1;
3571         }
3572
3573         ctdb->ctdbd_pid = getpid();
3574
3575         ctdb->recoverd_pid = fork();
3576         if (ctdb->recoverd_pid == -1) {
3577                 return -1;
3578         }
3579
3580         if (ctdb->recoverd_pid != 0) {
3581                 close(fd[0]);
3582                 event_add_timed(ctdb->ev, ctdb,
3583                                 timeval_current_ofs(30, 0),
3584                                 ctdb_check_recd, ctdb);
3585                 return 0;
3586         }
3587
3588         close(fd[1]);
3589
3590         srandom(getpid() ^ time(NULL));
3591
3592         if (switch_from_server_to_client(ctdb, "recoverd") != 0) {
3593                 DEBUG(DEBUG_CRIT, (__location__ "ERROR: failed to switch recovery daemon into client mode. shutting down.\n"));
3594                 exit(1);
3595         }
3596
3597         DEBUG(DEBUG_DEBUG, (__location__ " Created PIPE FD:%d to recovery daemon\n", fd[0]));
3598
3599         fde = event_add_fd(ctdb->ev, ctdb, fd[0], EVENT_FD_READ,
3600                      ctdb_recoverd_parent, &fd[0]);
3601         tevent_fd_set_auto_close(fde);
3602
3603         /* set up a handler to pick up sigchld */
3604         se = event_add_signal(ctdb->ev, ctdb,
3605                                      SIGCHLD, 0,
3606                                      recd_sig_child_handler,
3607                                      ctdb);
3608         if (se == NULL) {
3609                 DEBUG(DEBUG_CRIT,("Failed to set up signal handler for SIGCHLD in recovery daemon\n"));
3610                 exit(1);
3611         }
3612
3613         monitor_cluster(ctdb);
3614
3615         DEBUG(DEBUG_ALERT,("ERROR: ctdb_recoverd finished!?\n"));
3616         return -1;
3617 }
3618
3619 /*
3620   shutdown the recovery daemon
3621  */
3622 void ctdb_stop_recoverd(struct ctdb_context *ctdb)
3623 {
3624         if (ctdb->recoverd_pid == 0) {
3625                 return;
3626         }
3627
3628         DEBUG(DEBUG_NOTICE,("Shutting down recovery daemon\n"));
3629         kill(ctdb->recoverd_pid, SIGTERM);
3630 }
3631
3632 static void ctdb_restart_recd(struct event_context *ev, struct timed_event *te,
3633                        struct timeval t, void *private_data)
3634 {
3635         struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
3636
3637         DEBUG(DEBUG_ERR,("Restarting recovery daemon\n"));
3638         ctdb_stop_recoverd(ctdb);
3639         ctdb_start_recoverd(ctdb);
3640 }