ctdb/server/ctdb_recover.c

   1 /*
   2    ctdb recovery code
   3
   4    Copyright (C) Andrew Tridgell  2007
   5    Copyright (C) Ronnie Sahlberg  2007
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program; if not, see <http://www.gnu.org/licenses/>.
  19 */
  20 #include "replace.h"
  21 #include "system/time.h"
  22 #include "system/network.h"
  23 #include "system/filesys.h"
  24 #include "system/wait.h"
  25
  26 #include <talloc.h>
  27 #include <tevent.h>
  28 #include <tdb.h>
  29
  30 #include "lib/tdb_wrap/tdb_wrap.h"
  31 #include "lib/util/dlinklist.h"
  32 #include "lib/util/debug.h"
  33 #include "lib/util/samba_util.h"
  34
  35 #include "ctdb_private.h"
  36 #include "ctdb_client.h"
  37
  38 #include "common/system.h"
  39 #include "common/common.h"
  40 #include "common/logging.h"
  41
  42 int
  43 ctdb_control_getvnnmap(struct ctdb_context *ctdb, uint32_t opcode, TDB_DATA indata, TDB_DATA *outdata)
  44 {
  45         struct ctdb_vnn_map_wire *map;
  46         size_t len;
  47
  48         CHECK_CONTROL_DATA_SIZE(0);
  49
  50         len = offsetof(struct ctdb_vnn_map_wire, map) + sizeof(uint32_t)*ctdb->vnn_map->size;
  51         map = talloc_size(outdata, len);
  52         CTDB_NO_MEMORY(ctdb, map);
  53
  54         map->generation = ctdb->vnn_map->generation;
  55         map->size = ctdb->vnn_map->size;
  56         memcpy(map->map, ctdb->vnn_map->map, sizeof(uint32_t)*map->size);
  57
  58         outdata->dsize = len;
  59         outdata->dptr  = (uint8_t *)map;
  60
  61         return 0;
  62 }
  63
  64 int
  65 ctdb_control_setvnnmap(struct ctdb_context *ctdb, uint32_t opcode, TDB_DATA indata, TDB_DATA *outdata)
  66 {
  67         struct ctdb_vnn_map_wire *map = (struct ctdb_vnn_map_wire *)indata.dptr;
  68
  69         if (ctdb->recovery_mode != CTDB_RECOVERY_ACTIVE) {
  70                 DEBUG(DEBUG_ERR, ("Attempt to set vnnmap when not in recovery\n"));
  71                 return -1;
  72         }
  73
  74         talloc_free(ctdb->vnn_map);
  75
  76         ctdb->vnn_map = talloc(ctdb, struct ctdb_vnn_map);
  77         CTDB_NO_MEMORY(ctdb, ctdb->vnn_map);
  78
  79         ctdb->vnn_map->generation = map->generation;
  80         ctdb->vnn_map->size       = map->size;
  81         ctdb->vnn_map->map = talloc_array(ctdb->vnn_map, uint32_t, map->size);
  82         CTDB_NO_MEMORY(ctdb, ctdb->vnn_map->map);
  83
  84         memcpy(ctdb->vnn_map->map, map->map, sizeof(uint32_t)*map->size);
  85
  86         return 0;
  87 }
  88
  89 int
  90 ctdb_control_getdbmap(struct ctdb_context *ctdb, uint32_t opcode, TDB_DATA indata, TDB_DATA *outdata)
  91 {
  92         uint32_t i, len;
  93         struct ctdb_db_context *ctdb_db;
  94         struct ctdb_dbid_map_old *dbid_map;
  95
  96         CHECK_CONTROL_DATA_SIZE(0);
  97
  98         len = 0;
  99         for(ctdb_db=ctdb->db_list;ctdb_db;ctdb_db=ctdb_db->next){
 100                 len++;
 101         }
 102
 103
 104         outdata->dsize = offsetof(struct ctdb_dbid_map_old, dbs) + sizeof(dbid_map->dbs[0])*len;
 105         outdata->dptr  = (unsigned char *)talloc_zero_size(outdata, outdata->dsize);
 106         if (!outdata->dptr) {
 107                 DEBUG(DEBUG_ALERT, (__location__ " Failed to allocate dbmap array\n"));
 108                 exit(1);
 109         }
 110
 111         dbid_map = (struct ctdb_dbid_map_old *)outdata->dptr;
 112         dbid_map->num = len;
 113         for (i=0,ctdb_db=ctdb->db_list;ctdb_db;i++,ctdb_db=ctdb_db->next){
 114                 dbid_map->dbs[i].db_id       = ctdb_db->db_id;
 115                 if (ctdb_db->persistent != 0) {
 116                         dbid_map->dbs[i].flags |= CTDB_DB_FLAGS_PERSISTENT;
 117                 }
 118                 if (ctdb_db->readonly != 0) {
 119                         dbid_map->dbs[i].flags |= CTDB_DB_FLAGS_READONLY;
 120                 }
 121                 if (ctdb_db->sticky != 0) {
 122                         dbid_map->dbs[i].flags |= CTDB_DB_FLAGS_STICKY;
 123                 }
 124         }
 125
 126         return 0;
 127 }
 128
 129 int
 130 ctdb_control_getnodemap(struct ctdb_context *ctdb, uint32_t opcode, TDB_DATA indata, TDB_DATA *outdata)
 131 {
 132         CHECK_CONTROL_DATA_SIZE(0);
 133
 134         outdata->dptr  = (unsigned char *)ctdb_node_list_to_map(ctdb->nodes,
 135                                                                 ctdb->num_nodes,
 136                                                                 outdata);
 137         if (outdata->dptr == NULL) {
 138                 return -1;
 139         }
 140
 141         outdata->dsize = talloc_get_size(outdata->dptr);
 142
 143         return 0;
 144 }
 145
 146 /*
 147   reload the nodes file
 148 */
 149 int
 150 ctdb_control_reload_nodes_file(struct ctdb_context *ctdb, uint32_t opcode)
 151 {
 152         int i, num_nodes;
 153         TALLOC_CTX *tmp_ctx;
 154         struct ctdb_node **nodes;
 155
 156         tmp_ctx = talloc_new(ctdb);
 157
 158         /* steal the old nodes file for a while */
 159         talloc_steal(tmp_ctx, ctdb->nodes);
 160         nodes = ctdb->nodes;
 161         ctdb->nodes = NULL;
 162         num_nodes = ctdb->num_nodes;
 163         ctdb->num_nodes = 0;
 164
 165         /* load the new nodes file */
 166         ctdb_load_nodes_file(ctdb);
 167
 168         for (i=0; i<ctdb->num_nodes; i++) {
 169                 /* keep any identical pre-existing nodes and connections */
 170                 if ((i < num_nodes) && ctdb_same_address(&ctdb->nodes[i]->address, &nodes[i]->address)) {
 171                         talloc_free(ctdb->nodes[i]);
 172                         ctdb->nodes[i] = talloc_steal(ctdb->nodes, nodes[i]);
 173                         continue;
 174                 }
 175
 176                 if (ctdb->nodes[i]->flags & NODE_FLAGS_DELETED) {
 177                         continue;
 178                 }
 179
 180                 /* any new or different nodes must be added */
 181                 if (ctdb->methods->add_node(ctdb->nodes[i]) != 0) {
 182                         DEBUG(DEBUG_CRIT, (__location__ " methods->add_node failed at %d\n", i));
 183                         ctdb_fatal(ctdb, "failed to add node. shutting down\n");
 184                 }
 185                 if (ctdb->methods->connect_node(ctdb->nodes[i]) != 0) {
 186                         DEBUG(DEBUG_CRIT, (__location__ " methods->add_connect failed at %d\n", i));
 187                         ctdb_fatal(ctdb, "failed to connect to node. shutting down\n");
 188                 }
 189         }
 190
 191         /* tell the recovery daemon to reaload the nodes file too */
 192         ctdb_daemon_send_message(ctdb, ctdb->pnn, CTDB_SRVID_RELOAD_NODES, tdb_null);
 193
 194         talloc_free(tmp_ctx);
 195
 196         return 0;
 197 }
 198
 199 /*
 200    a traverse function for pulling all relevent records from pulldb
 201  */
 202 struct pulldb_data {
 203         struct ctdb_context *ctdb;
 204         struct ctdb_db_context *ctdb_db;
 205         struct ctdb_marshall_buffer *pulldata;
 206         uint32_t len;
 207         uint32_t allocated_len;
 208         bool failed;
 209 };
 210
 211 static int traverse_pulldb(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
 212 {
 213         struct pulldb_data *params = (struct pulldb_data *)p;
 214         struct ctdb_rec_data_old *rec;
 215         struct ctdb_context *ctdb = params->ctdb;
 216         struct ctdb_db_context *ctdb_db = params->ctdb_db;
 217
 218         /* add the record to the blob */
 219         rec = ctdb_marshall_record(params->pulldata, 0, key, NULL, data);
 220         if (rec == NULL) {
 221                 params->failed = true;
 222                 return -1;
 223         }
 224         if (params->len + rec->length >= params->allocated_len) {
 225                 params->allocated_len = rec->length + params->len + ctdb->tunable.pulldb_preallocation_size;
 226                 params->pulldata = talloc_realloc_size(NULL, params->pulldata, params->allocated_len);
 227         }
 228         if (params->pulldata == NULL) {
 229                 DEBUG(DEBUG_CRIT,(__location__ " Failed to expand pulldb_data to %u\n", rec->length + params->len));
 230                 ctdb_fatal(params->ctdb, "failed to allocate memory for recovery. shutting down\n");
 231         }
 232         params->pulldata->count++;
 233         memcpy(params->len+(uint8_t *)params->pulldata, rec, rec->length);
 234         params->len += rec->length;
 235
 236         if (ctdb->tunable.db_record_size_warn != 0 && rec->length > ctdb->tunable.db_record_size_warn) {
 237                 DEBUG(DEBUG_ERR,("Data record in %s is big. Record size is %d bytes\n", ctdb_db->db_name, (int)rec->length));
 238         }
 239
 240         talloc_free(rec);
 241
 242         return 0;
 243 }
 244
 245 /*
 246   pull a bunch of records from a ltdb, filtering by lmaster
 247  */
 248 int32_t ctdb_control_pull_db(struct ctdb_context *ctdb, TDB_DATA indata, TDB_DATA *outdata)
 249 {
 250         struct ctdb_pulldb *pull;
 251         struct ctdb_db_context *ctdb_db;
 252         struct pulldb_data params;
 253         struct ctdb_marshall_buffer *reply;
 254
 255         pull = (struct ctdb_pulldb *)indata.dptr;
 256
 257         ctdb_db = find_ctdb_db(ctdb, pull->db_id);
 258         if (!ctdb_db) {
 259                 DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%08x\n", pull->db_id));
 260                 return -1;
 261         }
 262
 263         if (!ctdb_db_frozen(ctdb_db)) {
 264                 DEBUG(DEBUG_ERR,
 265                       ("rejecting ctdb_control_pull_db when not frozen\n"));
 266                 return -1;
 267         }
 268
 269         reply = talloc_zero(outdata, struct ctdb_marshall_buffer);
 270         CTDB_NO_MEMORY(ctdb, reply);
 271
 272         reply->db_id = pull->db_id;
 273
 274         params.ctdb = ctdb;
 275         params.ctdb_db = ctdb_db;
 276         params.pulldata = reply;
 277         params.len = offsetof(struct ctdb_marshall_buffer, data);
 278         params.allocated_len = params.len;
 279         params.failed = false;
 280
 281         if (ctdb_db->unhealthy_reason) {
 282                 /* this is just a warning, as the tdb should be empty anyway */
 283                 DEBUG(DEBUG_WARNING,("db(%s) unhealty in ctdb_control_pull_db: %s\n",
 284                                      ctdb_db->db_name, ctdb_db->unhealthy_reason));
 285         }
 286
 287         if (ctdb_lockdb_mark(ctdb_db) != 0) {
 288                 DEBUG(DEBUG_ERR,(__location__ " Failed to get lock on entire db - failing\n"));
 289                 return -1;
 290         }
 291
 292         if (tdb_traverse_read(ctdb_db->ltdb->tdb, traverse_pulldb, &params) == -1) {
 293                 DEBUG(DEBUG_ERR,(__location__ " Failed to get traverse db '%s'\n", ctdb_db->db_name));
 294                 ctdb_lockdb_unmark(ctdb_db);
 295                 talloc_free(params.pulldata);
 296                 return -1;
 297         }
 298
 299         ctdb_lockdb_unmark(ctdb_db);
 300
 301         outdata->dptr = (uint8_t *)params.pulldata;
 302         outdata->dsize = params.len;
 303
 304         if (ctdb->tunable.db_record_count_warn != 0 && params.pulldata->count > ctdb->tunable.db_record_count_warn) {
 305                 DEBUG(DEBUG_ERR,("Database %s is big. Contains %d records\n", ctdb_db->db_name, params.pulldata->count));
 306         }
 307         if (ctdb->tunable.db_size_warn != 0 && outdata->dsize > ctdb->tunable.db_size_warn) {
 308                 DEBUG(DEBUG_ERR,("Database %s is big. Contains %d bytes\n", ctdb_db->db_name, (int)outdata->dsize));
 309         }
 310
 311
 312         return 0;
 313 }
 314
 315 /*
 316   push a bunch of records into a ltdb, filtering by rsn
 317  */
 318 int32_t ctdb_control_push_db(struct ctdb_context *ctdb, TDB_DATA indata)
 319 {
 320         struct ctdb_marshall_buffer *reply = (struct ctdb_marshall_buffer *)indata.dptr;
 321         struct ctdb_db_context *ctdb_db;
 322         int i, ret;
 323         struct ctdb_rec_data_old *rec;
 324
 325         if (indata.dsize < offsetof(struct ctdb_marshall_buffer, data)) {
 326                 DEBUG(DEBUG_ERR,(__location__ " invalid data in pulldb reply\n"));
 327                 return -1;
 328         }
 329
 330         ctdb_db = find_ctdb_db(ctdb, reply->db_id);
 331         if (!ctdb_db) {
 332                 DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%08x\n", reply->db_id));
 333                 return -1;
 334         }
 335
 336         if (!ctdb_db_frozen(ctdb_db)) {
 337                 DEBUG(DEBUG_ERR,
 338                       ("rejecting ctdb_control_push_db when not frozen\n"));
 339                 return -1;
 340         }
 341
 342         if (ctdb_lockdb_mark(ctdb_db) != 0) {
 343                 DEBUG(DEBUG_ERR,(__location__ " Failed to get lock on entire db - failing\n"));
 344                 return -1;
 345         }
 346
 347         rec = (struct ctdb_rec_data_old *)&reply->data[0];
 348
 349         DEBUG(DEBUG_INFO,("starting push of %u records for dbid 0x%x\n",
 350                  reply->count, reply->db_id));
 351
 352         for (i=0;i<reply->count;i++) {
 353                 TDB_DATA key, data;
 354                 struct ctdb_ltdb_header *hdr;
 355
 356                 key.dptr = &rec->data[0];
 357                 key.dsize = rec->keylen;
 358                 data.dptr = &rec->data[key.dsize];
 359                 data.dsize = rec->datalen;
 360
 361                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
 362                         DEBUG(DEBUG_CRIT,(__location__ " bad ltdb record\n"));
 363                         goto failed;
 364                 }
 365                 hdr = (struct ctdb_ltdb_header *)data.dptr;
 366                 /* strip off any read only record flags. All readonly records
 367                    are revoked implicitely by a recovery
 368                 */
 369                 hdr->flags &= ~CTDB_REC_RO_FLAGS;
 370
 371                 data.dptr += sizeof(*hdr);
 372                 data.dsize -= sizeof(*hdr);
 373
 374                 ret = ctdb_ltdb_store(ctdb_db, key, hdr, data);
 375                 if (ret != 0) {
 376                         DEBUG(DEBUG_CRIT, (__location__ " Unable to store record\n"));
 377                         goto failed;
 378                 }
 379
 380                 rec = (struct ctdb_rec_data_old *)(rec->length + (uint8_t *)rec);
 381         }
 382
 383         DEBUG(DEBUG_DEBUG,("finished push of %u records for dbid 0x%x\n",
 384                  reply->count, reply->db_id));
 385
 386         if (ctdb_db->readonly) {
 387                 DEBUG(DEBUG_CRIT,("Clearing the tracking database for dbid 0x%x\n",
 388                                   ctdb_db->db_id));
 389                 if (tdb_wipe_all(ctdb_db->rottdb) != 0) {
 390                         DEBUG(DEBUG_ERR,("Failed to wipe tracking database for 0x%x. Dropping read-only delegation support\n", ctdb_db->db_id));
 391                         ctdb_db->readonly = false;
 392                         tdb_close(ctdb_db->rottdb);
 393                         ctdb_db->rottdb = NULL;
 394                         ctdb_db->readonly = false;
 395                 }
 396                 while (ctdb_db->revokechild_active != NULL) {
 397                         talloc_free(ctdb_db->revokechild_active);
 398                 }
 399         }
 400
 401         ctdb_lockdb_unmark(ctdb_db);
 402         return 0;
 403
 404 failed:
 405         ctdb_lockdb_unmark(ctdb_db);
 406         return -1;
 407 }
 408
 409 struct ctdb_set_recmode_state {
 410         struct ctdb_context *ctdb;
 411         struct ctdb_req_control_old *c;
 412         uint32_t recmode;
 413         int fd[2];
 414         struct tevent_timer *te;
 415         struct tevent_fd *fde;
 416         pid_t child;
 417         struct timeval start_time;
 418 };
 419
 420 /*
 421   called if our set_recmode child times out. this would happen if
 422   ctdb_recovery_lock() would block.
 423  */
 424 static void ctdb_set_recmode_timeout(struct tevent_context *ev,
 425                                      struct tevent_timer *te,
 426                                      struct timeval t, void *private_data)
 427 {
 428         struct ctdb_set_recmode_state *state = talloc_get_type(private_data,
 429                                            struct ctdb_set_recmode_state);
 430
 431         /* we consider this a success, not a failure, as we failed to
 432            set the recovery lock which is what we wanted.  This can be
 433            caused by the cluster filesystem being very slow to
 434            arbitrate locks immediately after a node failure.
 435          */
 436         DEBUG(DEBUG_ERR,(__location__ " set_recmode child process hung/timedout CFS slow to grant locks? (allowing recmode set anyway)\n"));
 437         state->ctdb->recovery_mode = state->recmode;
 438         ctdb_request_control_reply(state->ctdb, state->c, NULL, 0, NULL);
 439         talloc_free(state);
 440 }
 441
 442
 443 /* when we free the recmode state we must kill any child process.
 444 */
 445 static int set_recmode_destructor(struct ctdb_set_recmode_state *state)
 446 {
 447         double l = timeval_elapsed(&state->start_time);
 448
 449         CTDB_UPDATE_RECLOCK_LATENCY(state->ctdb, "daemon reclock", reclock.ctdbd, l);
 450
 451         if (state->fd[0] != -1) {
 452                 state->fd[0] = -1;
 453         }
 454         if (state->fd[1] != -1) {
 455                 state->fd[1] = -1;
 456         }
 457         ctdb_kill(state->ctdb, state->child, SIGKILL);
 458         return 0;
 459 }
 460
 461 /* this is called when the client process has completed ctdb_recovery_lock()
 462    and has written data back to us through the pipe.
 463 */
 464 static void set_recmode_handler(struct tevent_context *ev,
 465                                 struct tevent_fd *fde,
 466                                 uint16_t flags, void *private_data)
 467 {
 468         struct ctdb_set_recmode_state *state= talloc_get_type(private_data,
 469                                              struct ctdb_set_recmode_state);
 470         char c = 0;
 471         int ret;
 472
 473         /* we got a response from our child process so we can abort the
 474            timeout.
 475         */
 476         talloc_free(state->te);
 477         state->te = NULL;
 478
 479
 480         /* If, as expected, the child was unable to take the recovery
 481          * lock then it will have written 0 into the pipe, so
 482          * continue.  However, any other value (e.g. 1) indicates that
 483          * it was able to take the recovery lock when it should have
 484          * been held by the recovery daemon on the recovery master.
 485         */
 486         ret = sys_read(state->fd[0], &c, 1);
 487         if (ret != 1 || c != 0) {
 488                 ctdb_request_control_reply(
 489                         state->ctdb, state->c, NULL, -1,
 490                         "Took recovery lock from daemon during recovery - probably a cluster filesystem lock coherence problem");
 491                 talloc_free(state);
 492                 return;
 493         }
 494
 495         state->ctdb->recovery_mode = state->recmode;
 496
 497         /* release any deferred attach calls from clients */
 498         if (state->recmode == CTDB_RECOVERY_NORMAL) {
 499                 ctdb_process_deferred_attach(state->ctdb);
 500         }
 501
 502         ctdb_request_control_reply(state->ctdb, state->c, NULL, 0, NULL);
 503         talloc_free(state);
 504         return;
 505 }
 506
 507 static void
 508 ctdb_drop_all_ips_event(struct tevent_context *ev, struct tevent_timer *te,
 509                         struct timeval t, void *private_data)
 510 {
 511         struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
 512
 513         DEBUG(DEBUG_ERR,(__location__ " Been in recovery mode for too long. Dropping all IPS\n"));
 514         talloc_free(ctdb->release_ips_ctx);
 515         ctdb->release_ips_ctx = NULL;
 516
 517         ctdb_release_all_ips(ctdb);
 518 }
 519
 520 /*
 521  * Set up an event to drop all public ips if we remain in recovery for too
 522  * long
 523  */
 524 int ctdb_deferred_drop_all_ips(struct ctdb_context *ctdb)
 525 {
 526         if (ctdb->release_ips_ctx != NULL) {
 527                 talloc_free(ctdb->release_ips_ctx);
 528         }
 529         ctdb->release_ips_ctx = talloc_new(ctdb);
 530         CTDB_NO_MEMORY(ctdb, ctdb->release_ips_ctx);
 531
 532         tevent_add_timer(ctdb->ev, ctdb->release_ips_ctx,
 533                          timeval_current_ofs(ctdb->tunable.recovery_drop_all_ips, 0),
 534                          ctdb_drop_all_ips_event, ctdb);
 535         return 0;
 536 }
 537
 538 /*
 539   set the recovery mode
 540  */
 541 int32_t ctdb_control_set_recmode(struct ctdb_context *ctdb,
 542                                  struct ctdb_req_control_old *c,
 543                                  TDB_DATA indata, bool *async_reply,
 544                                  const char **errormsg)
 545 {
 546         uint32_t recmode = *(uint32_t *)indata.dptr;
 547         int i, ret;
 548         struct ctdb_set_recmode_state *state;
 549         pid_t parent = getpid();
 550         struct ctdb_db_context *ctdb_db;
 551
 552         /* if we enter recovery but stay in recovery for too long
 553            we will eventually drop all our ip addresses
 554         */
 555         if (recmode == CTDB_RECOVERY_NORMAL) {
 556                 talloc_free(ctdb->release_ips_ctx);
 557                 ctdb->release_ips_ctx = NULL;
 558         } else {
 559                 if (ctdb_deferred_drop_all_ips(ctdb) != 0) {
 560                         DEBUG(DEBUG_ERR,("Failed to set up deferred drop all ips\n"));
 561                 }
 562         }
 563
 564         if (recmode != ctdb->recovery_mode) {
 565                 DEBUG(DEBUG_NOTICE,(__location__ " Recovery mode set to %s\n",
 566                          recmode==CTDB_RECOVERY_NORMAL?"NORMAL":"ACTIVE"));
 567         }
 568
 569         if (recmode != CTDB_RECOVERY_NORMAL ||
 570             ctdb->recovery_mode != CTDB_RECOVERY_ACTIVE) {
 571                 ctdb->recovery_mode = recmode;
 572                 return 0;
 573         }
 574
 575         /* some special handling when ending recovery mode */
 576
 577         for (ctdb_db = ctdb->db_list; ctdb_db != NULL; ctdb_db = ctdb_db->next) {
 578                 if (ctdb_db->generation != ctdb->vnn_map->generation) {
 579                         DEBUG(DEBUG_ERR,
 580                               ("Inconsistent DB generation %u for %s\n",
 581                                ctdb_db->generation, ctdb_db->db_name));
 582                         DEBUG(DEBUG_ERR, ("Recovery mode set to ACTIVE\n"));
 583                         return -1;
 584                 }
 585         }
 586
 587         /* force the databases to thaw */
 588         for (i=1; i<=NUM_DB_PRIORITIES; i++) {
 589                 if (ctdb_db_prio_frozen(ctdb, i)) {
 590                         ctdb_control_thaw(ctdb, i, false);
 591                 }
 592         }
 593
 594         state = talloc(ctdb, struct ctdb_set_recmode_state);
 595         CTDB_NO_MEMORY(ctdb, state);
 596
 597         state->start_time = timeval_current();
 598         state->fd[0] = -1;
 599         state->fd[1] = -1;
 600
 601         /* release any deferred attach calls from clients */
 602         if (recmode == CTDB_RECOVERY_NORMAL) {
 603                 ctdb_process_deferred_attach(ctdb);
 604         }
 605
 606         if (ctdb->recovery_lock_file == NULL) {
 607                 /* Not using recovery lock file */
 608                 ctdb->recovery_mode = recmode;
 609                 return 0;
 610         }
 611
 612         /* For the rest of what needs to be done, we need to do this in
 613            a child process since
 614            1, the call to ctdb_recovery_lock() can block if the cluster
 615               filesystem is in the process of recovery.
 616         */
 617         ret = pipe(state->fd);
 618         if (ret != 0) {
 619                 talloc_free(state);
 620                 DEBUG(DEBUG_CRIT,(__location__ " Failed to open pipe for set_recmode child\n"));
 621                 return -1;
 622         }
 623
 624         state->child = ctdb_fork(ctdb);
 625         if (state->child == (pid_t)-1) {
 626                 close(state->fd[0]);
 627                 close(state->fd[1]);
 628                 talloc_free(state);
 629                 return -1;
 630         }
 631
 632         if (state->child == 0) {
 633                 char cc = 0;
 634                 close(state->fd[0]);
 635
 636                 ctdb_set_process_name("ctdb_recmode");
 637                 debug_extra = talloc_asprintf(NULL, "set_recmode:");
 638                 /* Daemon should not be able to get the recover lock,
 639                  * as it should be held by the recovery master */
 640                 if (ctdb_recovery_lock(ctdb)) {
 641                         DEBUG(DEBUG_ERR,
 642                               ("ERROR: Daemon able to take recovery lock on \"%s\" during recovery\n",
 643                                ctdb->recovery_lock_file));
 644                         ctdb_recovery_unlock(ctdb);
 645                         cc = 1;
 646                 }
 647
 648                 sys_write(state->fd[1], &cc, 1);
 649                 /* make sure we die when our parent dies */
 650                 while (ctdb_kill(ctdb, parent, 0) == 0 || errno != ESRCH) {
 651                         sleep(5);
 652                         sys_write(state->fd[1], &cc, 1);
 653                 }
 654                 _exit(0);
 655         }
 656         close(state->fd[1]);
 657         set_close_on_exec(state->fd[0]);
 658
 659         state->fd[1] = -1;
 660
 661         talloc_set_destructor(state, set_recmode_destructor);
 662
 663         DEBUG(DEBUG_DEBUG, (__location__ " Created PIPE FD:%d for setrecmode\n", state->fd[0]));
 664
 665         state->te = tevent_add_timer(ctdb->ev, state, timeval_current_ofs(5, 0),
 666                                      ctdb_set_recmode_timeout, state);
 667
 668         state->fde = tevent_add_fd(ctdb->ev, state, state->fd[0], TEVENT_FD_READ,
 669                                    set_recmode_handler, (void *)state);
 670
 671         if (state->fde == NULL) {
 672                 talloc_free(state);
 673                 return -1;
 674         }
 675         tevent_fd_set_auto_close(state->fde);
 676
 677         state->ctdb    = ctdb;
 678         state->recmode = recmode;
 679         state->c       = talloc_steal(state, c);
 680
 681         *async_reply = true;
 682
 683         return 0;
 684 }
 685
 686
 687 bool ctdb_recovery_have_lock(struct ctdb_context *ctdb)
 688 {
 689         return ctdb->recovery_lock_fd != -1;
 690 }
 691
 692 /*
 693   try and get the recovery lock in shared storage - should only work
 694   on the recovery master recovery daemon. Anywhere else is a bug
 695  */
 696 bool ctdb_recovery_lock(struct ctdb_context *ctdb)
 697 {
 698         struct flock lock;
 699
 700         ctdb->recovery_lock_fd = open(ctdb->recovery_lock_file,
 701                                       O_RDWR|O_CREAT, 0600);
 702         if (ctdb->recovery_lock_fd == -1) {
 703                 DEBUG(DEBUG_ERR,
 704                       ("ctdb_recovery_lock: Unable to open %s - (%s)\n",
 705                        ctdb->recovery_lock_file, strerror(errno)));
 706                 return false;
 707         }
 708
 709         set_close_on_exec(ctdb->recovery_lock_fd);
 710
 711         lock.l_type = F_WRLCK;
 712         lock.l_whence = SEEK_SET;
 713         lock.l_start = 0;
 714         lock.l_len = 1;
 715         lock.l_pid = 0;
 716
 717         if (fcntl(ctdb->recovery_lock_fd, F_SETLK, &lock) != 0) {
 718                 int saved_errno = errno;
 719                 close(ctdb->recovery_lock_fd);
 720                 ctdb->recovery_lock_fd = -1;
 721                 /* Fail silently on these errors, since they indicate
 722                  * lock contention, but log an error for any other
 723                  * failure. */
 724                 if (saved_errno != EACCES &&
 725                     saved_errno != EAGAIN) {
 726                         DEBUG(DEBUG_ERR,("ctdb_recovery_lock: Failed to get "
 727                                          "recovery lock on '%s' - (%s)\n",
 728                                          ctdb->recovery_lock_file,
 729                                          strerror(saved_errno)));
 730                 }
 731                 return false;
 732         }
 733
 734         return true;
 735 }
 736
 737 void ctdb_recovery_unlock(struct ctdb_context *ctdb)
 738 {
 739         if (ctdb->recovery_lock_fd != -1) {
 740                 DEBUG(DEBUG_NOTICE, ("Releasing recovery lock\n"));
 741                 close(ctdb->recovery_lock_fd);
 742                 ctdb->recovery_lock_fd = -1;
 743         }
 744 }
 745
 746 /*
 747   delete a record as part of the vacuum process
 748   only delete if we are not lmaster or dmaster, and our rsn is <= the provided rsn
 749   use non-blocking locks
 750
 751   return 0 if the record was successfully deleted (i.e. it does not exist
 752   when the function returns)
 753   or !0 is the record still exists in the tdb after returning.
 754  */
 755 static int delete_tdb_record(struct ctdb_context *ctdb, struct ctdb_db_context *ctdb_db, struct ctdb_rec_data_old *rec)
 756 {
 757         TDB_DATA key, data, data2;
 758         struct ctdb_ltdb_header *hdr, *hdr2;
 759
 760         /* these are really internal tdb functions - but we need them here for
 761            non-blocking lock of the freelist */
 762         int tdb_lock_nonblock(struct tdb_context *tdb, int list, int ltype);
 763         int tdb_unlock(struct tdb_context *tdb, int list, int ltype);
 764
 765
 766         key.dsize = rec->keylen;
 767         key.dptr  = &rec->data[0];
 768         data.dsize = rec->datalen;
 769         data.dptr = &rec->data[rec->keylen];
 770
 771         if (ctdb_lmaster(ctdb, &key) == ctdb->pnn) {
 772                 DEBUG(DEBUG_INFO,(__location__ " Called delete on record where we are lmaster\n"));
 773                 return -1;
 774         }
 775
 776         if (data.dsize != sizeof(struct ctdb_ltdb_header)) {
 777                 DEBUG(DEBUG_ERR,(__location__ " Bad record size\n"));
 778                 return -1;
 779         }
 780
 781         hdr = (struct ctdb_ltdb_header *)data.dptr;
 782
 783         /* use a non-blocking lock */
 784         if (tdb_chainlock_nonblock(ctdb_db->ltdb->tdb, key) != 0) {
 785                 return -1;
 786         }
 787
 788         data2 = tdb_fetch(ctdb_db->ltdb->tdb, key);
 789         if (data2.dptr == NULL) {
 790                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 791                 return 0;
 792         }
 793
 794         if (data2.dsize < sizeof(struct ctdb_ltdb_header)) {
 795                 if (tdb_lock_nonblock(ctdb_db->ltdb->tdb, -1, F_WRLCK) == 0) {
 796                         if (tdb_delete(ctdb_db->ltdb->tdb, key) != 0) {
 797                                 DEBUG(DEBUG_CRIT,(__location__ " Failed to delete corrupt record\n"));
 798                         }
 799                         tdb_unlock(ctdb_db->ltdb->tdb, -1, F_WRLCK);
 800                         DEBUG(DEBUG_CRIT,(__location__ " Deleted corrupt record\n"));
 801                 }
 802                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 803                 free(data2.dptr);
 804                 return 0;
 805         }
 806
 807         hdr2 = (struct ctdb_ltdb_header *)data2.dptr;
 808
 809         if (hdr2->rsn > hdr->rsn) {
 810                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 811                 DEBUG(DEBUG_INFO,(__location__ " Skipping record with rsn=%llu - called with rsn=%llu\n",
 812                          (unsigned long long)hdr2->rsn, (unsigned long long)hdr->rsn));
 813                 free(data2.dptr);
 814                 return -1;
 815         }
 816
 817         /* do not allow deleting record that have readonly flags set. */
 818         if (hdr->flags & CTDB_REC_RO_FLAGS) {
 819                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 820                 DEBUG(DEBUG_INFO,(__location__ " Skipping record with readonly flags set\n"));
 821                 free(data2.dptr);
 822                 return -1;
 823         }
 824         if (hdr2->flags & CTDB_REC_RO_FLAGS) {
 825                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 826                 DEBUG(DEBUG_INFO,(__location__ " Skipping record with readonly flags set\n"));
 827                 free(data2.dptr);
 828                 return -1;
 829         }
 830
 831         if (hdr2->dmaster == ctdb->pnn) {
 832                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 833                 DEBUG(DEBUG_INFO,(__location__ " Attempted delete record where we are the dmaster\n"));
 834                 free(data2.dptr);
 835                 return -1;
 836         }
 837
 838         if (tdb_lock_nonblock(ctdb_db->ltdb->tdb, -1, F_WRLCK) != 0) {
 839                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 840                 free(data2.dptr);
 841                 return -1;
 842         }
 843
 844         if (tdb_delete(ctdb_db->ltdb->tdb, key) != 0) {
 845                 tdb_unlock(ctdb_db->ltdb->tdb, -1, F_WRLCK);
 846                 tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 847                 DEBUG(DEBUG_INFO,(__location__ " Failed to delete record\n"));
 848                 free(data2.dptr);
 849                 return -1;
 850         }
 851
 852         tdb_unlock(ctdb_db->ltdb->tdb, -1, F_WRLCK);
 853         tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 854         free(data2.dptr);
 855         return 0;
 856 }
 857
 858
 859
 860 struct recovery_callback_state {
 861         struct ctdb_req_control_old *c;
 862 };
 863
 864
 865 /*
 866   called when the 'recovered' event script has finished
 867  */
 868 static void ctdb_end_recovery_callback(struct ctdb_context *ctdb, int status, void *p)
 869 {
 870         struct recovery_callback_state *state = talloc_get_type(p, struct recovery_callback_state);
 871
 872         ctdb_enable_monitoring(ctdb);
 873         CTDB_INCREMENT_STAT(ctdb, num_recoveries);
 874
 875         if (status != 0) {
 876                 DEBUG(DEBUG_ERR,(__location__ " recovered event script failed (status %d)\n", status));
 877                 if (status == -ETIME) {
 878                         ctdb_ban_self(ctdb);
 879                 }
 880         }
 881
 882         ctdb_request_control_reply(ctdb, state->c, NULL, status, NULL);
 883         talloc_free(state);
 884
 885         gettimeofday(&ctdb->last_recovery_finished, NULL);
 886
 887         if (ctdb->runstate == CTDB_RUNSTATE_FIRST_RECOVERY) {
 888                 ctdb_set_runstate(ctdb, CTDB_RUNSTATE_STARTUP);
 889         }
 890 }
 891
 892 /*
 893   recovery has finished
 894  */
 895 int32_t ctdb_control_end_recovery(struct ctdb_context *ctdb,
 896                                 struct ctdb_req_control_old *c,
 897                                 bool *async_reply)
 898 {
 899         int ret;
 900         struct recovery_callback_state *state;
 901
 902         DEBUG(DEBUG_NOTICE,("Recovery has finished\n"));
 903
 904         ctdb_persistent_finish_trans3_commits(ctdb);
 905
 906         state = talloc(ctdb, struct recovery_callback_state);
 907         CTDB_NO_MEMORY(ctdb, state);
 908
 909         state->c    = c;
 910
 911         ctdb_disable_monitoring(ctdb);
 912
 913         ret = ctdb_event_script_callback(ctdb, state,
 914                                          ctdb_end_recovery_callback,
 915                                          state,
 916                                          CTDB_EVENT_RECOVERED, "%s", "");
 917
 918         if (ret != 0) {
 919                 ctdb_enable_monitoring(ctdb);
 920
 921                 DEBUG(DEBUG_ERR,(__location__ " Failed to end recovery\n"));
 922                 talloc_free(state);
 923                 return -1;
 924         }
 925
 926         /* tell the control that we will be reply asynchronously */
 927         state->c    = talloc_steal(state, c);
 928         *async_reply = true;
 929         return 0;
 930 }
 931
 932 /*
 933   called when the 'startrecovery' event script has finished
 934  */
 935 static void ctdb_start_recovery_callback(struct ctdb_context *ctdb, int status, void *p)
 936 {
 937         struct recovery_callback_state *state = talloc_get_type(p, struct recovery_callback_state);
 938
 939         if (status != 0) {
 940                 DEBUG(DEBUG_ERR,(__location__ " startrecovery event script failed (status %d)\n", status));
 941         }
 942
 943         ctdb_request_control_reply(ctdb, state->c, NULL, status, NULL);
 944         talloc_free(state);
 945 }
 946
 947 /*
 948   run the startrecovery eventscript
 949  */
 950 int32_t ctdb_control_start_recovery(struct ctdb_context *ctdb,
 951                                 struct ctdb_req_control_old *c,
 952                                 bool *async_reply)
 953 {
 954         int ret;
 955         struct recovery_callback_state *state;
 956
 957         DEBUG(DEBUG_NOTICE,(__location__ " startrecovery eventscript has been invoked\n"));
 958         gettimeofday(&ctdb->last_recovery_started, NULL);
 959
 960         state = talloc(ctdb, struct recovery_callback_state);
 961         CTDB_NO_MEMORY(ctdb, state);
 962
 963         state->c    = talloc_steal(state, c);
 964
 965         ctdb_disable_monitoring(ctdb);
 966
 967         ret = ctdb_event_script_callback(ctdb, state,
 968                                          ctdb_start_recovery_callback,
 969                                          state,
 970                                          CTDB_EVENT_START_RECOVERY,
 971                                          "%s", "");
 972
 973         if (ret != 0) {
 974                 DEBUG(DEBUG_ERR,(__location__ " Failed to start recovery\n"));
 975                 talloc_free(state);
 976                 return -1;
 977         }
 978
 979         /* tell the control that we will be reply asynchronously */
 980         *async_reply = true;
 981         return 0;
 982 }
 983
 984 /*
 985  try to delete all these records as part of the vacuuming process
 986  and return the records we failed to delete
 987 */
 988 int32_t ctdb_control_try_delete_records(struct ctdb_context *ctdb, TDB_DATA indata, TDB_DATA *outdata)
 989 {
 990         struct ctdb_marshall_buffer *reply = (struct ctdb_marshall_buffer *)indata.dptr;
 991         struct ctdb_db_context *ctdb_db;
 992         int i;
 993         struct ctdb_rec_data_old *rec;
 994         struct ctdb_marshall_buffer *records;
 995
 996         if (indata.dsize < offsetof(struct ctdb_marshall_buffer, data)) {
 997                 DEBUG(DEBUG_ERR,(__location__ " invalid data in try_delete_records\n"));
 998                 return -1;
 999         }
1000
1001         ctdb_db = find_ctdb_db(ctdb, reply->db_id);
1002         if (!ctdb_db) {
1003                 DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%08x\n", reply->db_id));
1004                 return -1;
1005         }
1006
1007
1008         DEBUG(DEBUG_DEBUG,("starting try_delete_records of %u records for dbid 0x%x\n",
1009                  reply->count, reply->db_id));
1010
1011
1012         /* create a blob to send back the records we couldnt delete */
1013         records = (struct ctdb_marshall_buffer *)
1014                         talloc_zero_size(outdata,
1015                                     offsetof(struct ctdb_marshall_buffer, data));
1016         if (records == NULL) {
1017                 DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
1018                 return -1;
1019         }
1020         records->db_id = ctdb_db->db_id;
1021
1022
1023         rec = (struct ctdb_rec_data_old *)&reply->data[0];
1024         for (i=0;i<reply->count;i++) {
1025                 TDB_DATA key, data;
1026
1027                 key.dptr = &rec->data[0];
1028                 key.dsize = rec->keylen;
1029                 data.dptr = &rec->data[key.dsize];
1030                 data.dsize = rec->datalen;
1031
1032                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1033                         DEBUG(DEBUG_CRIT,(__location__ " bad ltdb record in indata\n"));
1034                         return -1;
1035                 }
1036
1037                 /* If we cant delete the record we must add it to the reply
1038                    so the lmaster knows it may not purge this record
1039                 */
1040                 if (delete_tdb_record(ctdb, ctdb_db, rec) != 0) {
1041                         size_t old_size;
1042                         struct ctdb_ltdb_header *hdr;
1043
1044                         hdr = (struct ctdb_ltdb_header *)data.dptr;
1045                         data.dptr += sizeof(*hdr);
1046                         data.dsize -= sizeof(*hdr);
1047
1048                         DEBUG(DEBUG_INFO, (__location__ " Failed to vacuum delete record with hash 0x%08x\n", ctdb_hash(&key)));
1049
1050                         old_size = talloc_get_size(records);
1051                         records = talloc_realloc_size(outdata, records, old_size + rec->length);
1052                         if (records == NULL) {
1053                                 DEBUG(DEBUG_ERR,(__location__ " Failed to expand\n"));
1054                                 return -1;
1055                         }
1056                         records->count++;
1057                         memcpy(old_size+(uint8_t *)records, rec, rec->length);
1058                 }
1059
1060                 rec = (struct ctdb_rec_data_old *)(rec->length + (uint8_t *)rec);
1061         }
1062
1063
1064         *outdata = ctdb_marshall_finish(records);
1065
1066         return 0;
1067 }
1068
1069 /**
1070  * Store a record as part of the vacuum process:
1071  * This is called from the RECEIVE_RECORD control which
1072  * the lmaster uses to send the current empty copy
1073  * to all nodes for storing, before it lets the other
1074  * nodes delete the records in the second phase with
1075  * the TRY_DELETE_RECORDS control.
1076  *
1077  * Only store if we are not lmaster or dmaster, and our
1078  * rsn is <= the provided rsn. Use non-blocking locks.
1079  *
1080  * return 0 if the record was successfully stored.
1081  * return !0 if the record still exists in the tdb after returning.
1082  */
1083 static int store_tdb_record(struct ctdb_context *ctdb,
1084                             struct ctdb_db_context *ctdb_db,
1085                             struct ctdb_rec_data_old *rec)
1086 {
1087         TDB_DATA key, data, data2;
1088         struct ctdb_ltdb_header *hdr, *hdr2;
1089         int ret;
1090
1091         key.dsize = rec->keylen;
1092         key.dptr = &rec->data[0];
1093         data.dsize = rec->datalen;
1094         data.dptr = &rec->data[rec->keylen];
1095
1096         if (ctdb_lmaster(ctdb, &key) == ctdb->pnn) {
1097                 DEBUG(DEBUG_INFO, (__location__ " Called store_tdb_record "
1098                                    "where we are lmaster\n"));
1099                 return -1;
1100         }
1101
1102         if (data.dsize != sizeof(struct ctdb_ltdb_header)) {
1103                 DEBUG(DEBUG_ERR, (__location__ " Bad record size\n"));
1104                 return -1;
1105         }
1106
1107         hdr = (struct ctdb_ltdb_header *)data.dptr;
1108
1109         /* use a non-blocking lock */
1110         if (tdb_chainlock_nonblock(ctdb_db->ltdb->tdb, key) != 0) {
1111                 DEBUG(DEBUG_INFO, (__location__ " Failed to lock chain in non-blocking mode\n"));
1112                 return -1;
1113         }
1114
1115         data2 = tdb_fetch(ctdb_db->ltdb->tdb, key);
1116         if (data2.dptr == NULL || data2.dsize < sizeof(struct ctdb_ltdb_header)) {
1117                 if (tdb_store(ctdb_db->ltdb->tdb, key, data, 0) == -1) {
1118                         DEBUG(DEBUG_ERR, (__location__ "Failed to store record\n"));
1119                         ret = -1;
1120                         goto done;
1121                 }
1122                 DEBUG(DEBUG_INFO, (__location__ " Stored record\n"));
1123                 ret = 0;
1124                 goto done;
1125         }
1126
1127         hdr2 = (struct ctdb_ltdb_header *)data2.dptr;
1128
1129         if (hdr2->rsn > hdr->rsn) {
1130                 DEBUG(DEBUG_INFO, (__location__ " Skipping record with "
1131                                    "rsn=%llu - called with rsn=%llu\n",
1132                                    (unsigned long long)hdr2->rsn,
1133                                    (unsigned long long)hdr->rsn));
1134                 ret = -1;
1135                 goto done;
1136         }
1137
1138         /* do not allow vacuuming of records that have readonly flags set. */
1139         if (hdr->flags & CTDB_REC_RO_FLAGS) {
1140                 DEBUG(DEBUG_INFO,(__location__ " Skipping record with readonly "
1141                                   "flags set\n"));
1142                 ret = -1;
1143                 goto done;
1144         }
1145         if (hdr2->flags & CTDB_REC_RO_FLAGS) {
1146                 DEBUG(DEBUG_INFO,(__location__ " Skipping record with readonly "
1147                                   "flags set\n"));
1148                 ret = -1;
1149                 goto done;
1150         }
1151
1152         if (hdr2->dmaster == ctdb->pnn) {
1153                 DEBUG(DEBUG_INFO, (__location__ " Attempted to store record "
1154                                    "where we are the dmaster\n"));
1155                 ret = -1;
1156                 goto done;
1157         }
1158
1159         if (tdb_store(ctdb_db->ltdb->tdb, key, data, 0) != 0) {
1160                 DEBUG(DEBUG_INFO,(__location__ " Failed to store record\n"));
1161                 ret = -1;
1162                 goto done;
1163         }
1164
1165         ret = 0;
1166
1167 done:
1168         tdb_chainunlock(ctdb_db->ltdb->tdb, key);
1169         free(data2.dptr);
1170         return  ret;
1171 }
1172
1173
1174
1175 /**
1176  * Try to store all these records as part of the vacuuming process
1177  * and return the records we failed to store.
1178  */
1179 int32_t ctdb_control_receive_records(struct ctdb_context *ctdb,
1180                                      TDB_DATA indata, TDB_DATA *outdata)
1181 {
1182         struct ctdb_marshall_buffer *reply = (struct ctdb_marshall_buffer *)indata.dptr;
1183         struct ctdb_db_context *ctdb_db;
1184         int i;
1185         struct ctdb_rec_data_old *rec;
1186         struct ctdb_marshall_buffer *records;
1187
1188         if (indata.dsize < offsetof(struct ctdb_marshall_buffer, data)) {
1189                 DEBUG(DEBUG_ERR,
1190                       (__location__ " invalid data in receive_records\n"));
1191                 return -1;
1192         }
1193
1194         ctdb_db = find_ctdb_db(ctdb, reply->db_id);
1195         if (!ctdb_db) {
1196                 DEBUG(DEBUG_ERR, (__location__ " Unknown db 0x%08x\n",
1197                                   reply->db_id));
1198                 return -1;
1199         }
1200
1201         DEBUG(DEBUG_DEBUG, ("starting receive_records of %u records for "
1202                             "dbid 0x%x\n", reply->count, reply->db_id));
1203
1204         /* create a blob to send back the records we could not store */
1205         records = (struct ctdb_marshall_buffer *)
1206                         talloc_zero_size(outdata,
1207                                 offsetof(struct ctdb_marshall_buffer, data));
1208         if (records == NULL) {
1209                 DEBUG(DEBUG_ERR, (__location__ " Out of memory\n"));
1210                 return -1;
1211         }
1212         records->db_id = ctdb_db->db_id;
1213
1214         rec = (struct ctdb_rec_data_old *)&reply->data[0];
1215         for (i=0; i<reply->count; i++) {
1216                 TDB_DATA key, data;
1217
1218                 key.dptr = &rec->data[0];
1219                 key.dsize = rec->keylen;
1220                 data.dptr = &rec->data[key.dsize];
1221                 data.dsize = rec->datalen;
1222
1223                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1224                         DEBUG(DEBUG_CRIT, (__location__ " bad ltdb record "
1225                                            "in indata\n"));
1226                         return -1;
1227                 }
1228
1229                 /*
1230                  * If we can not store the record we must add it to the reply
1231                  * so the lmaster knows it may not purge this record.
1232                  */
1233                 if (store_tdb_record(ctdb, ctdb_db, rec) != 0) {
1234                         size_t old_size;
1235                         struct ctdb_ltdb_header *hdr;
1236
1237                         hdr = (struct ctdb_ltdb_header *)data.dptr;
1238                         data.dptr += sizeof(*hdr);
1239                         data.dsize -= sizeof(*hdr);
1240
1241                         DEBUG(DEBUG_INFO, (__location__ " Failed to store "
1242                                            "record with hash 0x%08x in vacuum "
1243                                            "via RECEIVE_RECORDS\n",
1244                                            ctdb_hash(&key)));
1245
1246                         old_size = talloc_get_size(records);
1247                         records = talloc_realloc_size(outdata, records,
1248                                                       old_size + rec->length);
1249                         if (records == NULL) {
1250                                 DEBUG(DEBUG_ERR, (__location__ " Failed to "
1251                                                   "expand\n"));
1252                                 return -1;
1253                         }
1254                         records->count++;
1255                         memcpy(old_size+(uint8_t *)records, rec, rec->length);
1256                 }
1257
1258                 rec = (struct ctdb_rec_data_old *)(rec->length + (uint8_t *)rec);
1259         }
1260
1261         *outdata = ctdb_marshall_finish(records);
1262
1263         return 0;
1264 }
1265
1266
1267 /*
1268   report capabilities
1269  */
1270 int32_t ctdb_control_get_capabilities(struct ctdb_context *ctdb, TDB_DATA *outdata)
1271 {
1272         uint32_t *capabilities = NULL;
1273
1274         capabilities = talloc(outdata, uint32_t);
1275         CTDB_NO_MEMORY(ctdb, capabilities);
1276         *capabilities = ctdb->capabilities;
1277
1278         outdata->dsize = sizeof(uint32_t);
1279         outdata->dptr = (uint8_t *)capabilities;
1280
1281         return 0;
1282 }
1283
1284 /* The recovery daemon will ping us at regular intervals.
1285    If we havent been pinged for a while we assume the recovery
1286    daemon is inoperable and we restart.
1287 */
1288 static void ctdb_recd_ping_timeout(struct tevent_context *ev,
1289                                    struct tevent_timer *te,
1290                                    struct timeval t, void *p)
1291 {
1292         struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
1293         uint32_t *count = talloc_get_type(ctdb->recd_ping_count, uint32_t);
1294
1295         DEBUG(DEBUG_ERR, ("Recovery daemon ping timeout. Count : %u\n", *count));
1296
1297         if (*count < ctdb->tunable.recd_ping_failcount) {
1298                 (*count)++;
1299                 tevent_add_timer(ctdb->ev, ctdb->recd_ping_count,
1300                                  timeval_current_ofs(ctdb->tunable.recd_ping_timeout, 0),
1301                                  ctdb_recd_ping_timeout, ctdb);
1302                 return;
1303         }
1304
1305         DEBUG(DEBUG_ERR, ("Final timeout for recovery daemon ping. Restarting recovery daemon. (This can be caused if the cluster filesystem has hung)\n"));
1306
1307         ctdb_stop_recoverd(ctdb);
1308         ctdb_start_recoverd(ctdb);
1309 }
1310
1311 int32_t ctdb_control_recd_ping(struct ctdb_context *ctdb)
1312 {
1313         talloc_free(ctdb->recd_ping_count);
1314
1315         ctdb->recd_ping_count = talloc_zero(ctdb, uint32_t);
1316         CTDB_NO_MEMORY(ctdb, ctdb->recd_ping_count);
1317
1318         if (ctdb->tunable.recd_ping_timeout != 0) {
1319                 tevent_add_timer(ctdb->ev, ctdb->recd_ping_count,
1320                                  timeval_current_ofs(ctdb->tunable.recd_ping_timeout, 0),
1321                                  ctdb_recd_ping_timeout, ctdb);
1322         }
1323
1324         return 0;
1325 }
1326
1327
1328
1329 int32_t ctdb_control_set_recmaster(struct ctdb_context *ctdb, uint32_t opcode, TDB_DATA indata)
1330 {
1331         uint32_t new_recmaster;
1332
1333         CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
1334         new_recmaster = ((uint32_t *)(&indata.dptr[0]))[0];
1335
1336         if (ctdb->pnn != new_recmaster && ctdb->recovery_master == ctdb->pnn) {
1337                 DEBUG(DEBUG_NOTICE,
1338                       ("This node (%u) is no longer the recovery master\n", ctdb->pnn));
1339         }
1340
1341         if (ctdb->pnn == new_recmaster && ctdb->recovery_master != new_recmaster) {
1342                 DEBUG(DEBUG_NOTICE,
1343                       ("This node (%u) is now the recovery master\n", ctdb->pnn));
1344         }
1345
1346         ctdb->recovery_master = new_recmaster;
1347         return 0;
1348 }
1349
1350
1351 int32_t ctdb_control_stop_node(struct ctdb_context *ctdb)
1352 {
1353         DEBUG(DEBUG_NOTICE, ("Stopping node\n"));
1354         ctdb_disable_monitoring(ctdb);
1355         ctdb->nodes[ctdb->pnn]->flags |= NODE_FLAGS_STOPPED;
1356
1357         return 0;
1358 }
1359
1360 int32_t ctdb_control_continue_node(struct ctdb_context *ctdb)
1361 {
1362         DEBUG(DEBUG_NOTICE, ("Continue node\n"));
1363         ctdb->nodes[ctdb->pnn]->flags &= ~NODE_FLAGS_STOPPED;
1364
1365         return 0;
1366 }
1367