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