?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
ctdb-recovery: Ban a node that causes recovery failure
[Samba.git] / ctdb / server / ctdb_recovery_helper.c
blob7fdcc2e5a29ea0414bb8fe84a44e6fa9415b3cba
1 /*
2 ctdb parallel database recovery
3
4 Copyright (C) Amitay Isaacs 2015
5
6 This program is free software; you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
10
11 This program is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
15
16 You should have received a copy of the GNU General Public License
17 along with this program; if not, see <http://www.gnu.org/licenses/>.
18 */
19
20 #include "replace.h"
21 #include "system/network.h"
22 #include "system/filesys.h"
23
24 #include <talloc.h>
25 #include <tevent.h>
26 #include <tdb.h>
27 #include <libgen.h>
28
29 #include "lib/tdb_wrap/tdb_wrap.h"
30 #include "lib/util/sys_rw.h"
31 #include "lib/util/time.h"
32 #include "lib/util/tevent_unix.h"
33
34 #include "protocol/protocol.h"
35 #include "protocol/protocol_api.h"
36 #include "client/client.h"
37
38 #include "common/logging.h"
39
40 static int recover_timeout = 30;
41
42 #define NUM_RETRIES 3
43
44 #define TIMEOUT() timeval_current_ofs(recover_timeout, 0)
45
46 /*
47 * Utility functions
48 */
49
50 static bool generic_recv(struct tevent_req *req, int *perr)
51 {
52 int err;
53
54 if (tevent_req_is_unix_error(req, &err)) {
55 if (perr != NULL) {
56 *perr = err;
57 }
58 return false;
59 }
60
61 return true;
62 }
63
64 static uint64_t rec_srvid = CTDB_SRVID_RECOVERY;
65
66 static uint64_t srvid_next(void)
67 {
68 rec_srvid += 1;
69 return rec_srvid;
70 }
71
72 /*
73 * Recovery database functions
74 */
75
76 struct recdb_context {
77 uint32_t db_id;
78 const char *db_name;
79 const char *db_path;
80 struct tdb_wrap *db;
81 bool persistent;
82 };
83
84 static struct recdb_context *recdb_create(TALLOC_CTX *mem_ctx, uint32_t db_id,
85 const char *db_name,
86 const char *db_path,
87 uint32_t hash_size, bool persistent)
88 {
89 static char *db_dir_state = NULL;
90 struct recdb_context *recdb;
91 unsigned int tdb_flags;
92
93 recdb = talloc(mem_ctx, struct recdb_context);
94 if (recdb == NULL) {
95 return NULL;
96 }
97
98 if (db_dir_state == NULL) {
99 db_dir_state = getenv("CTDB_DBDIR_STATE");
100 }
101
102 recdb->db_name = db_name;
103 recdb->db_id = db_id;
104 recdb->db_path = talloc_asprintf(recdb, "%s/recdb.%s",
105 db_dir_state != NULL ?
106 db_dir_state :
107 dirname(discard_const(db_path)),
108 db_name);
109 if (recdb->db_path == NULL) {
110 talloc_free(recdb);
111 return NULL;
112 }
113 unlink(recdb->db_path);
114
115 tdb_flags = TDB_NOLOCK | TDB_INCOMPATIBLE_HASH | TDB_DISALLOW_NESTING;
116 recdb->db = tdb_wrap_open(mem_ctx, recdb->db_path, hash_size,
117 tdb_flags, O_RDWR|O_CREAT|O_EXCL, 0600);
118 if (recdb->db == NULL) {
119 talloc_free(recdb);
120 D_ERR("failed to create recovery db %s\n", recdb->db_path);
121 return NULL;
122 }
123
124 recdb->persistent = persistent;
125
126 return recdb;
127 }
128
129 static uint32_t recdb_id(struct recdb_context *recdb)
130 {
131 return recdb->db_id;
132 }
133
134 static const char *recdb_name(struct recdb_context *recdb)
135 {
136 return recdb->db_name;
137 }
138
139 static const char *recdb_path(struct recdb_context *recdb)
140 {
141 return recdb->db_path;
142 }
143
144 static struct tdb_context *recdb_tdb(struct recdb_context *recdb)
145 {
146 return recdb->db->tdb;
147 }
148
149 static bool recdb_persistent(struct recdb_context *recdb)
150 {
151 return recdb->persistent;
152 }
153
154 struct recdb_add_traverse_state {
155 struct recdb_context *recdb;
156 int mypnn;
157 };
158
159 static int recdb_add_traverse(uint32_t reqid, struct ctdb_ltdb_header *header,
160 TDB_DATA key, TDB_DATA data,
161 void *private_data)
162 {
163 struct recdb_add_traverse_state *state =
164 (struct recdb_add_traverse_state *)private_data;
165 struct ctdb_ltdb_header *hdr;
166 TDB_DATA prev_data;
167 int ret;
168
169 /* header is not marshalled separately in the pulldb control */
170 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
171 return -1;
172 }
173
174 hdr = (struct ctdb_ltdb_header *)data.dptr;
175
176 /* fetch the existing record, if any */
177 prev_data = tdb_fetch(recdb_tdb(state->recdb), key);
178
179 if (prev_data.dptr != NULL) {
180 struct ctdb_ltdb_header prev_hdr;
181
182 prev_hdr = *(struct ctdb_ltdb_header *)prev_data.dptr;
183 free(prev_data.dptr);
184 if (hdr->rsn < prev_hdr.rsn ||
185 (hdr->rsn == prev_hdr.rsn &&
186 prev_hdr.dmaster != state->mypnn)) {
187 return 0;
188 }
189 }
190
191 ret = tdb_store(recdb_tdb(state->recdb), key, data, TDB_REPLACE);
192 if (ret != 0) {
193 return -1;
194 }
195 return 0;
196 }
197
198 static bool recdb_add(struct recdb_context *recdb, int mypnn,
199 struct ctdb_rec_buffer *recbuf)
200 {
201 struct recdb_add_traverse_state state;
202 int ret;
203
204 state.recdb = recdb;
205 state.mypnn = mypnn;
206
207 ret = ctdb_rec_buffer_traverse(recbuf, recdb_add_traverse, &state);
208 if (ret != 0) {
209 return false;
210 }
211
212 return true;
213 }
214
215 /* This function decides which records from recdb are retained */
216 static int recbuf_filter_add(struct ctdb_rec_buffer *recbuf, bool persistent,
217 uint32_t reqid, uint32_t dmaster,
218 TDB_DATA key, TDB_DATA data)
219 {
220 struct ctdb_ltdb_header *header;
221 int ret;
222
223 /* Skip empty records */
224 if (data.dsize <= sizeof(struct ctdb_ltdb_header)) {
225 return 0;
226 }
227
228 /* update the dmaster field to point to us */
229 header = (struct ctdb_ltdb_header *)data.dptr;
230 if (!persistent) {
231 header->dmaster = dmaster;
232 header->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
233 }
234
235 ret = ctdb_rec_buffer_add(recbuf, recbuf, reqid, NULL, key, data);
236 if (ret != 0) {
237 return ret;
238 }
239
240 return 0;
241 }
242
243 struct recdb_records_traverse_state {
244 struct ctdb_rec_buffer *recbuf;
245 uint32_t dmaster;
246 uint32_t reqid;
247 bool persistent;
248 bool failed;
249 };
250
251 static int recdb_records_traverse(struct tdb_context *tdb,
252 TDB_DATA key, TDB_DATA data,
253 void *private_data)
254 {
255 struct recdb_records_traverse_state *state =
256 (struct recdb_records_traverse_state *)private_data;
257 int ret;
258
259 ret = recbuf_filter_add(state->recbuf, state->persistent,
260 state->reqid, state->dmaster, key, data);
261 if (ret != 0) {
262 state->failed = true;
263 return ret;
264 }
265
266 return 0;
267 }
268
269 static struct ctdb_rec_buffer *recdb_records(struct recdb_context *recdb,
270 TALLOC_CTX *mem_ctx,
271 uint32_t dmaster)
272 {
273 struct recdb_records_traverse_state state;
274 int ret;
275
276 state.recbuf = ctdb_rec_buffer_init(mem_ctx, recdb_id(recdb));
277 if (state.recbuf == NULL) {
278 return NULL;
279 }
280 state.dmaster = dmaster;
281 state.reqid = 0;
282 state.persistent = recdb_persistent(recdb);
283 state.failed = false;
284
285 ret = tdb_traverse_read(recdb_tdb(recdb), recdb_records_traverse,
286 &state);
287 if (ret == -1 || state.failed) {
288 D_ERR("Failed to marshall recovery records for %s\n",
289 recdb_name(recdb));
290 TALLOC_FREE(state.recbuf);
291 return NULL;
292 }
293
294 return state.recbuf;
295 }
296
297 struct recdb_file_traverse_state {
298 struct ctdb_rec_buffer *recbuf;
299 struct recdb_context *recdb;
300 TALLOC_CTX *mem_ctx;
301 uint32_t dmaster;
302 uint32_t reqid;
303 bool persistent;
304 bool failed;
305 int fd;
306 int max_size;
307 int num_buffers;
308 };
309
310 static int recdb_file_traverse(struct tdb_context *tdb,
311 TDB_DATA key, TDB_DATA data,
312 void *private_data)
313 {
314 struct recdb_file_traverse_state *state =
315 (struct recdb_file_traverse_state *)private_data;
316 int ret;
317
318 ret = recbuf_filter_add(state->recbuf, state->persistent,
319 state->reqid, state->dmaster, key, data);
320 if (ret != 0) {
321 state->failed = true;
322 return ret;
323 }
324
325 if (ctdb_rec_buffer_len(state->recbuf) > state->max_size) {
326 ret = ctdb_rec_buffer_write(state->recbuf, state->fd);
327 if (ret != 0) {
328 D_ERR("Failed to collect recovery records for %s\n",
329 recdb_name(state->recdb));
330 state->failed = true;
331 return ret;
332 }
333
334 state->num_buffers += 1;
335
336 TALLOC_FREE(state->recbuf);
337 state->recbuf = ctdb_rec_buffer_init(state->mem_ctx,
338 recdb_id(state->recdb));
339 if (state->recbuf == NULL) {
340 state->failed = true;
341 return ENOMEM;
342 }
343 }
344
345 return 0;
346 }
347
348 static int recdb_file(struct recdb_context *recdb, TALLOC_CTX *mem_ctx,
349 uint32_t dmaster, int fd, int max_size)
350 {
351 struct recdb_file_traverse_state state;
352 int ret;
353
354 state.recbuf = ctdb_rec_buffer_init(mem_ctx, recdb_id(recdb));
355 if (state.recbuf == NULL) {
356 return -1;
357 }
358 state.recdb = recdb;
359 state.mem_ctx = mem_ctx;
360 state.dmaster = dmaster;
361 state.reqid = 0;
362 state.persistent = recdb_persistent(recdb);
363 state.failed = false;
364 state.fd = fd;
365 state.max_size = max_size;
366 state.num_buffers = 0;
367
368 ret = tdb_traverse_read(recdb_tdb(recdb), recdb_file_traverse, &state);
369 if (ret == -1 || state.failed) {
370 TALLOC_FREE(state.recbuf);
371 return -1;
372 }
373
374 ret = ctdb_rec_buffer_write(state.recbuf, fd);
375 if (ret != 0) {
376 D_ERR("Failed to collect recovery records for %s\n",
377 recdb_name(recdb));
378 TALLOC_FREE(state.recbuf);
379 return -1;
380 }
381 state.num_buffers += 1;
382
383 D_DEBUG("Wrote %d buffers of recovery records for %s\n",
384 state.num_buffers, recdb_name(recdb));
385
386 return state.num_buffers;
387 }
388
389 /*
390 * Pull database from a single node
391 */
392
393 struct pull_database_state {
394 struct tevent_context *ev;
395 struct ctdb_client_context *client;
396 struct recdb_context *recdb;
397 uint32_t pnn;
398 uint64_t srvid;
399 int num_records;
400 int result;
401 };
402
403 static void pull_database_handler(uint64_t srvid, TDB_DATA data,
404 void *private_data);
405 static void pull_database_register_done(struct tevent_req *subreq);
406 static void pull_database_old_done(struct tevent_req *subreq);
407 static void pull_database_unregister_done(struct tevent_req *subreq);
408 static void pull_database_new_done(struct tevent_req *subreq);
409
410 static struct tevent_req *pull_database_send(
411 TALLOC_CTX *mem_ctx,
412 struct tevent_context *ev,
413 struct ctdb_client_context *client,
414 uint32_t pnn, uint32_t caps,
415 struct recdb_context *recdb)
416 {
417 struct tevent_req *req, *subreq;
418 struct pull_database_state *state;
419 struct ctdb_req_control request;
420
421 req = tevent_req_create(mem_ctx, &state, struct pull_database_state);
422 if (req == NULL) {
423 return NULL;
424 }
425
426 state->ev = ev;
427 state->client = client;
428 state->recdb = recdb;
429 state->pnn = pnn;
430 state->srvid = srvid_next();
431
432 if (caps & CTDB_CAP_FRAGMENTED_CONTROLS) {
433 subreq = ctdb_client_set_message_handler_send(
434 state, state->ev, state->client,
435 state->srvid, pull_database_handler,
436 req);
437 if (tevent_req_nomem(subreq, req)) {
438 return tevent_req_post(req, ev);
439 }
440
441 tevent_req_set_callback(subreq, pull_database_register_done,
442 req);
443
444 } else {
445 struct ctdb_pulldb pulldb;
446
447 pulldb.db_id = recdb_id(recdb);
448 pulldb.lmaster = CTDB_LMASTER_ANY;
449
450 ctdb_req_control_pull_db(&request, &pulldb);
451 subreq = ctdb_client_control_send(state, state->ev,
452 state->client,
453 pnn, TIMEOUT(),
454 &request);
455 if (tevent_req_nomem(subreq, req)) {
456 return tevent_req_post(req, ev);
457 }
458 tevent_req_set_callback(subreq, pull_database_old_done, req);
459 }
460
461 return req;
462 }
463
464 static void pull_database_handler(uint64_t srvid, TDB_DATA data,
465 void *private_data)
466 {
467 struct tevent_req *req = talloc_get_type_abort(
468 private_data, struct tevent_req);
469 struct pull_database_state *state = tevent_req_data(
470 req, struct pull_database_state);
471 struct ctdb_rec_buffer *recbuf;
472 size_t np;
473 int ret;
474 bool status;
475
476 if (srvid != state->srvid) {
477 return;
478 }
479
480 ret = ctdb_rec_buffer_pull(data.dptr, data.dsize, state, &recbuf, &np);
481 if (ret != 0) {
482 D_ERR("Invalid data received for DB_PULL messages\n");
483 return;
484 }
485
486 if (recbuf->db_id != recdb_id(state->recdb)) {
487 talloc_free(recbuf);
488 D_ERR("Invalid dbid:%08x for DB_PULL messages for %s\n",
489 recbuf->db_id, recdb_name(state->recdb));
490 return;
491 }
492
493 status = recdb_add(state->recdb, ctdb_client_pnn(state->client),
494 recbuf);
495 if (! status) {
496 talloc_free(recbuf);
497 D_ERR("Failed to add records to recdb for %s\n",
498 recdb_name(state->recdb));
499 return;
500 }
501
502 state->num_records += recbuf->count;
503 talloc_free(recbuf);
504 }
505
506 static void pull_database_register_done(struct tevent_req *subreq)
507 {
508 struct tevent_req *req = tevent_req_callback_data(
509 subreq, struct tevent_req);
510 struct pull_database_state *state = tevent_req_data(
511 req, struct pull_database_state);
512 struct ctdb_req_control request;
513 struct ctdb_pulldb_ext pulldb_ext;
514 int ret;
515 bool status;
516
517 status = ctdb_client_set_message_handler_recv(subreq, &ret);
518 TALLOC_FREE(subreq);
519 if (! status) {
520 D_ERR("Failed to set message handler for DB_PULL for %s\n",
521 recdb_name(state->recdb));
522 tevent_req_error(req, ret);
523 return;
524 }
525
526 pulldb_ext.db_id = recdb_id(state->recdb);
527 pulldb_ext.lmaster = CTDB_LMASTER_ANY;
528 pulldb_ext.srvid = state->srvid;
529
530 ctdb_req_control_db_pull(&request, &pulldb_ext);
531 subreq = ctdb_client_control_send(state, state->ev, state->client,
532 state->pnn, TIMEOUT(), &request);
533 if (tevent_req_nomem(subreq, req)) {
534 return;
535 }
536 tevent_req_set_callback(subreq, pull_database_new_done, req);
537 }
538
539 static void pull_database_old_done(struct tevent_req *subreq)
540 {
541 struct tevent_req *req = tevent_req_callback_data(
542 subreq, struct tevent_req);
543 struct pull_database_state *state = tevent_req_data(
544 req, struct pull_database_state);
545 struct ctdb_reply_control *reply;
546 struct ctdb_rec_buffer *recbuf;
547 int ret;
548 bool status;
549
550 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
551 TALLOC_FREE(subreq);
552 if (! status) {
553 D_ERR("control PULL_DB failed for %s on node %u, ret=%d\n",
554 recdb_name(state->recdb), state->pnn, ret);
555 tevent_req_error(req, ret);
556 return;
557 }
558
559 ret = ctdb_reply_control_pull_db(reply, state, &recbuf);
560 talloc_free(reply);
561 if (ret != 0) {
562 tevent_req_error(req, ret);
563 return;
564 }
565
566 status = recdb_add(state->recdb, ctdb_client_pnn(state->client),
567 recbuf);
568 if (! status) {
569 talloc_free(recbuf);
570 tevent_req_error(req, EIO);
571 return;
572 }
573
574 state->num_records = recbuf->count;
575 talloc_free(recbuf);
576
577 D_INFO("Pulled %d records for db %s from node %d\n",
578 state->num_records, recdb_name(state->recdb), state->pnn);
579
580 tevent_req_done(req);
581 }
582
583 static void pull_database_new_done(struct tevent_req *subreq)
584 {
585 struct tevent_req *req = tevent_req_callback_data(
586 subreq, struct tevent_req);
587 struct pull_database_state *state = tevent_req_data(
588 req, struct pull_database_state);
589 struct ctdb_reply_control *reply;
590 uint32_t num_records;
591 int ret;
592 bool status;
593
594 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
595 TALLOC_FREE(subreq);
596 if (! status) {
597 D_ERR("control DB_PULL failed for %s on node %u, ret=%d\n",
598 recdb_name(state->recdb), state->pnn, ret);
599 state->result = ret;
600 goto unregister;
601 }
602
603 ret = ctdb_reply_control_db_pull(reply, &num_records);
604 talloc_free(reply);
605 if (num_records != state->num_records) {
606 D_ERR("mismatch (%u != %u) in DB_PULL records for db %s\n",
607 num_records, state->num_records,
608 recdb_name(state->recdb));
609 state->result = EIO;
610 goto unregister;
611 }
612
613 D_INFO("Pulled %d records for db %s from node %d\n",
614 state->num_records, recdb_name(state->recdb), state->pnn);
615
616 unregister:
617
618 subreq = ctdb_client_remove_message_handler_send(
619 state, state->ev, state->client,
620 state->srvid, req);
621 if (tevent_req_nomem(subreq, req)) {
622 return;
623 }
624 tevent_req_set_callback(subreq, pull_database_unregister_done, req);
625 }
626
627 static void pull_database_unregister_done(struct tevent_req *subreq)
628 {
629 struct tevent_req *req = tevent_req_callback_data(
630 subreq, struct tevent_req);
631 struct pull_database_state *state = tevent_req_data(
632 req, struct pull_database_state);
633 int ret;
634 bool status;
635
636 status = ctdb_client_remove_message_handler_recv(subreq, &ret);
637 TALLOC_FREE(subreq);
638 if (! status) {
639 D_ERR("failed to remove message handler for DB_PULL for db %s\n",
640 recdb_name(state->recdb));
641 tevent_req_error(req, ret);
642 return;
643 }
644
645 if (state->result != 0) {
646 tevent_req_error(req, state->result);
647 return;
648 }
649
650 tevent_req_done(req);
651 }
652
653 static bool pull_database_recv(struct tevent_req *req, int *perr)
654 {
655 return generic_recv(req, perr);
656 }
657
658 /*
659 * Push database to specified nodes (old style)
660 */
661
662 struct push_database_old_state {
663 struct tevent_context *ev;
664 struct ctdb_client_context *client;
665 struct recdb_context *recdb;
666 uint32_t *pnn_list;
667 int count;
668 struct ctdb_rec_buffer *recbuf;
669 int index;
670 };
671
672 static void push_database_old_push_done(struct tevent_req *subreq);
673
674 static struct tevent_req *push_database_old_send(
675 TALLOC_CTX *mem_ctx,
676 struct tevent_context *ev,
677 struct ctdb_client_context *client,
678 uint32_t *pnn_list, int count,
679 struct recdb_context *recdb)
680 {
681 struct tevent_req *req, *subreq;
682 struct push_database_old_state *state;
683 struct ctdb_req_control request;
684 uint32_t pnn;
685
686 req = tevent_req_create(mem_ctx, &state,
687 struct push_database_old_state);
688 if (req == NULL) {
689 return NULL;
690 }
691
692 state->ev = ev;
693 state->client = client;
694 state->recdb = recdb;
695 state->pnn_list = pnn_list;
696 state->count = count;
697 state->index = 0;
698
699 state->recbuf = recdb_records(recdb, state,
700 ctdb_client_pnn(client));
701 if (tevent_req_nomem(state->recbuf, req)) {
702 return tevent_req_post(req, ev);
703 }
704
705 pnn = state->pnn_list[state->index];
706
707 ctdb_req_control_push_db(&request, state->recbuf);
708 subreq = ctdb_client_control_send(state, ev, client, pnn,
709 TIMEOUT(), &request);
710 if (tevent_req_nomem(subreq, req)) {
711 return tevent_req_post(req, ev);
712 }
713 tevent_req_set_callback(subreq, push_database_old_push_done, req);
714
715 return req;
716 }
717
718 static void push_database_old_push_done(struct tevent_req *subreq)
719 {
720 struct tevent_req *req = tevent_req_callback_data(
721 subreq, struct tevent_req);
722 struct push_database_old_state *state = tevent_req_data(
723 req, struct push_database_old_state);
724 struct ctdb_req_control request;
725 uint32_t pnn;
726 int ret;
727 bool status;
728
729 status = ctdb_client_control_recv(subreq, &ret, NULL, NULL);
730 TALLOC_FREE(subreq);
731 if (! status) {
732 D_ERR("control PUSH_DB failed for db %s on node %u, ret=%d\n",
733 recdb_name(state->recdb), state->pnn_list[state->index],
734 ret);
735 tevent_req_error(req, ret);
736 return;
737 }
738
739 state->index += 1;
740 if (state->index == state->count) {
741 TALLOC_FREE(state->recbuf);
742 tevent_req_done(req);
743 return;
744 }
745
746 pnn = state->pnn_list[state->index];
747
748 ctdb_req_control_push_db(&request, state->recbuf);
749 subreq = ctdb_client_control_send(state, state->ev, state->client,
750 pnn, TIMEOUT(), &request);
751 if (tevent_req_nomem(subreq, req)) {
752 return;
753 }
754 tevent_req_set_callback(subreq, push_database_old_push_done, req);
755 }
756
757 static bool push_database_old_recv(struct tevent_req *req, int *perr)
758 {
759 return generic_recv(req, perr);
760 }
761
762 /*
763 * Push database to specified nodes (new style)
764 */
765
766 struct push_database_new_state {
767 struct tevent_context *ev;
768 struct ctdb_client_context *client;
769 struct recdb_context *recdb;
770 uint32_t *pnn_list;
771 int count;
772 uint64_t srvid;
773 uint32_t dmaster;
774 int fd;
775 int num_buffers;
776 int num_buffers_sent;
777 int num_records;
778 };
779
780 static void push_database_new_started(struct tevent_req *subreq);
781 static void push_database_new_send_msg(struct tevent_req *req);
782 static void push_database_new_send_done(struct tevent_req *subreq);
783 static void push_database_new_confirmed(struct tevent_req *subreq);
784
785 static struct tevent_req *push_database_new_send(
786 TALLOC_CTX *mem_ctx,
787 struct tevent_context *ev,
788 struct ctdb_client_context *client,
789 uint32_t *pnn_list, int count,
790 struct recdb_context *recdb,
791 int max_size)
792 {
793 struct tevent_req *req, *subreq;
794 struct push_database_new_state *state;
795 struct ctdb_req_control request;
796 struct ctdb_pulldb_ext pulldb_ext;
797 char *filename;
798 off_t offset;
799
800 req = tevent_req_create(mem_ctx, &state,
801 struct push_database_new_state);
802 if (req == NULL) {
803 return NULL;
804 }
805
806 state->ev = ev;
807 state->client = client;
808 state->recdb = recdb;
809 state->pnn_list = pnn_list;
810 state->count = count;
811
812 state->srvid = srvid_next();
813 state->dmaster = ctdb_client_pnn(client);
814 state->num_buffers_sent = 0;
815 state->num_records = 0;
816
817 filename = talloc_asprintf(state, "%s.dat", recdb_path(recdb));
818 if (tevent_req_nomem(filename, req)) {
819 return tevent_req_post(req, ev);
820 }
821
822 state->fd = open(filename, O_RDWR|O_CREAT, 0644);
823 if (state->fd == -1) {
824 tevent_req_error(req, errno);
825 return tevent_req_post(req, ev);
826 }
827 unlink(filename);
828 talloc_free(filename);
829
830 state->num_buffers = recdb_file(recdb, state, state->dmaster,
831 state->fd, max_size);
832 if (state->num_buffers == -1) {
833 tevent_req_error(req, ENOMEM);
834 return tevent_req_post(req, ev);
835 }
836
837 offset = lseek(state->fd, 0, SEEK_SET);
838 if (offset != 0) {
839 tevent_req_error(req, EIO);
840 return tevent_req_post(req, ev);
841 }
842
843 pulldb_ext.db_id = recdb_id(recdb);
844 pulldb_ext.srvid = state->srvid;
845
846 ctdb_req_control_db_push_start(&request, &pulldb_ext);
847 subreq = ctdb_client_control_multi_send(state, ev, client,
848 pnn_list, count,
849 TIMEOUT(), &request);
850 if (tevent_req_nomem(subreq, req)) {
851 return tevent_req_post(req, ev);
852 }
853 tevent_req_set_callback(subreq, push_database_new_started, req);
854
855 return req;
856 }
857
858 static void push_database_new_started(struct tevent_req *subreq)
859 {
860 struct tevent_req *req = tevent_req_callback_data(
861 subreq, struct tevent_req);
862 struct push_database_new_state *state = tevent_req_data(
863 req, struct push_database_new_state);
864 int *err_list;
865 int ret;
866 bool status;
867
868 status = ctdb_client_control_multi_recv(subreq, &ret, state,
869 &err_list, NULL);
870 TALLOC_FREE(subreq);
871 if (! status) {
872 int ret2;
873 uint32_t pnn;
874
875 ret2 = ctdb_client_control_multi_error(state->pnn_list,
876 state->count,
877 err_list, &pnn);
878 if (ret2 != 0) {
879 D_ERR("control DB_PUSH_START failed for db %s"
880 " on node %u, ret=%d\n",
881 recdb_name(state->recdb), pnn, ret2);
882 } else {
883 D_ERR("control DB_PUSH_START failed for db %s,"
884 " ret=%d\n",
885 recdb_name(state->recdb), ret);
886 }
887 talloc_free(err_list);
888
889 tevent_req_error(req, ret);
890 return;
891 }
892
893 push_database_new_send_msg(req);
894 }
895
896 static void push_database_new_send_msg(struct tevent_req *req)
897 {
898 struct push_database_new_state *state = tevent_req_data(
899 req, struct push_database_new_state);
900 struct tevent_req *subreq;
901 struct ctdb_rec_buffer *recbuf;
902 struct ctdb_req_message message;
903 TDB_DATA data;
904 size_t np;
905 int ret;
906
907 if (state->num_buffers_sent == state->num_buffers) {
908 struct ctdb_req_control request;
909
910 ctdb_req_control_db_push_confirm(&request,
911 recdb_id(state->recdb));
912 subreq = ctdb_client_control_multi_send(state, state->ev,
913 state->client,
914 state->pnn_list,
915 state->count,
916 TIMEOUT(), &request);
917 if (tevent_req_nomem(subreq, req)) {
918 return;
919 }
920 tevent_req_set_callback(subreq, push_database_new_confirmed,
921 req);
922 return;
923 }
924
925 ret = ctdb_rec_buffer_read(state->fd, state, &recbuf);
926 if (ret != 0) {
927 tevent_req_error(req, ret);
928 return;
929 }
930
931 data.dsize = ctdb_rec_buffer_len(recbuf);
932 data.dptr = talloc_size(state, data.dsize);
933 if (tevent_req_nomem(data.dptr, req)) {
934 return;
935 }
936
937 ctdb_rec_buffer_push(recbuf, data.dptr, &np);
938
939 message.srvid = state->srvid;
940 message.data.data = data;
941
942 D_DEBUG("Pushing buffer %d with %d records for db %s\n",
943 state->num_buffers_sent, recbuf->count,
944 recdb_name(state->recdb));
945
946 subreq = ctdb_client_message_multi_send(state, state->ev,
947 state->client,
948 state->pnn_list, state->count,
949 &message);
950 if (tevent_req_nomem(subreq, req)) {
951 return;
952 }
953 tevent_req_set_callback(subreq, push_database_new_send_done, req);
954
955 state->num_records += recbuf->count;
956
957 talloc_free(data.dptr);
958 talloc_free(recbuf);
959 }
960
961 static void push_database_new_send_done(struct tevent_req *subreq)
962 {
963 struct tevent_req *req = tevent_req_callback_data(
964 subreq, struct tevent_req);
965 struct push_database_new_state *state = tevent_req_data(
966 req, struct push_database_new_state);
967 bool status;
968 int ret;
969
970 status = ctdb_client_message_multi_recv(subreq, &ret, NULL, NULL);
971 TALLOC_FREE(subreq);
972 if (! status) {
973 D_ERR("Sending recovery records failed for %s\n",
974 recdb_name(state->recdb));
975 tevent_req_error(req, ret);
976 return;
977 }
978
979 state->num_buffers_sent += 1;
980
981 push_database_new_send_msg(req);
982 }
983
984 static void push_database_new_confirmed(struct tevent_req *subreq)
985 {
986 struct tevent_req *req = tevent_req_callback_data(
987 subreq, struct tevent_req);
988 struct push_database_new_state *state = tevent_req_data(
989 req, struct push_database_new_state);
990 struct ctdb_reply_control **reply;
991 int *err_list;
992 bool status;
993 int ret, i;
994 uint32_t num_records;
995
996 status = ctdb_client_control_multi_recv(subreq, &ret, state,
997 &err_list, &reply);
998 TALLOC_FREE(subreq);
999 if (! status) {
1000 int ret2;
1001 uint32_t pnn;
1002
1003 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1004 state->count, err_list,
1005 &pnn);
1006 if (ret2 != 0) {
1007 D_ERR("control DB_PUSH_CONFIRM failed for db %s"
1008 " on node %u, ret=%d\n",
1009 recdb_name(state->recdb), pnn, ret2);
1010 } else {
1011 D_ERR("control DB_PUSH_CONFIRM failed for db %s,"
1012 " ret=%d\n",
1013 recdb_name(state->recdb), ret);
1014 }
1015 tevent_req_error(req, ret);
1016 return;
1017 }
1018
1019 for (i=0; i<state->count; i++) {
1020 ret = ctdb_reply_control_db_push_confirm(reply[i],
1021 &num_records);
1022 if (ret != 0) {
1023 tevent_req_error(req, EPROTO);
1024 return;
1025 }
1026
1027 if (num_records != state->num_records) {
1028 D_ERR("Node %u received %d of %d records for %s\n",
1029 state->pnn_list[i], num_records,
1030 state->num_records, recdb_name(state->recdb));
1031 tevent_req_error(req, EPROTO);
1032 return;
1033 }
1034 }
1035
1036 talloc_free(reply);
1037
1038 D_INFO("Pushed %d records for db %s\n",
1039 state->num_records, recdb_name(state->recdb));
1040
1041 tevent_req_done(req);
1042 }
1043
1044 static bool push_database_new_recv(struct tevent_req *req, int *perr)
1045 {
1046 return generic_recv(req, perr);
1047 }
1048
1049 /*
1050 * wrapper for push_database_old and push_database_new
1051 */
1052
1053 struct push_database_state {
1054 bool old_done, new_done;
1055 };
1056
1057 static void push_database_old_done(struct tevent_req *subreq);
1058 static void push_database_new_done(struct tevent_req *subreq);
1059
1060 static struct tevent_req *push_database_send(
1061 TALLOC_CTX *mem_ctx,
1062 struct tevent_context *ev,
1063 struct ctdb_client_context *client,
1064 uint32_t *pnn_list, int count, uint32_t *caps,
1065 struct ctdb_tunable_list *tun_list,
1066 struct recdb_context *recdb)
1067 {
1068 struct tevent_req *req, *subreq;
1069 struct push_database_state *state;
1070 uint32_t *old_list, *new_list;
1071 unsigned int old_count, new_count;
1072 int i;
1073
1074 req = tevent_req_create(mem_ctx, &state, struct push_database_state);
1075 if (req == NULL) {
1076 return NULL;
1077 }
1078
1079 state->old_done = false;
1080 state->new_done = false;
1081
1082 old_count = 0;
1083 new_count = 0;
1084 old_list = talloc_array(state, uint32_t, count);
1085 new_list = talloc_array(state, uint32_t, count);
1086 if (tevent_req_nomem(old_list, req) ||
1087 tevent_req_nomem(new_list,req)) {
1088 return tevent_req_post(req, ev);
1089 }
1090
1091 for (i=0; i<count; i++) {
1092 uint32_t pnn = pnn_list[i];
1093
1094 if (caps[pnn] & CTDB_CAP_FRAGMENTED_CONTROLS) {
1095 new_list[new_count] = pnn;
1096 new_count += 1;
1097 } else {
1098 old_list[old_count] = pnn;
1099 old_count += 1;
1100 }
1101 }
1102
1103 if (old_count > 0) {
1104 subreq = push_database_old_send(state, ev, client,
1105 old_list, old_count, recdb);
1106 if (tevent_req_nomem(subreq, req)) {
1107 return tevent_req_post(req, ev);
1108 }
1109 tevent_req_set_callback(subreq, push_database_old_done, req);
1110 } else {
1111 state->old_done = true;
1112 }
1113
1114 if (new_count > 0) {
1115 subreq = push_database_new_send(state, ev, client,
1116 new_list, new_count, recdb,
1117 tun_list->rec_buffer_size_limit);
1118 if (tevent_req_nomem(subreq, req)) {
1119 return tevent_req_post(req, ev);
1120 }
1121 tevent_req_set_callback(subreq, push_database_new_done, req);
1122 } else {
1123 state->new_done = true;
1124 }
1125
1126 return req;
1127 }
1128
1129 static void push_database_old_done(struct tevent_req *subreq)
1130 {
1131 struct tevent_req *req = tevent_req_callback_data(
1132 subreq, struct tevent_req);
1133 struct push_database_state *state = tevent_req_data(
1134 req, struct push_database_state);
1135 bool status;
1136 int ret;
1137
1138 status = push_database_old_recv(subreq, &ret);
1139 if (! status) {
1140 tevent_req_error(req, ret);
1141 return;
1142 }
1143
1144 state->old_done = true;
1145
1146 if (state->old_done && state->new_done) {
1147 tevent_req_done(req);
1148 }
1149 }
1150
1151 static void push_database_new_done(struct tevent_req *subreq)
1152 {
1153 struct tevent_req *req = tevent_req_callback_data(
1154 subreq, struct tevent_req);
1155 struct push_database_state *state = tevent_req_data(
1156 req, struct push_database_state);
1157 bool status;
1158 int ret;
1159
1160 status = push_database_new_recv(subreq, &ret);
1161 if (! status) {
1162 tevent_req_error(req, ret);
1163 return;
1164 }
1165
1166 state->new_done = true;
1167
1168 if (state->old_done && state->new_done) {
1169 tevent_req_done(req);
1170 }
1171 }
1172
1173 static bool push_database_recv(struct tevent_req *req, int *perr)
1174 {
1175 return generic_recv(req, perr);
1176 }
1177
1178 /*
1179 * Collect databases using highest sequence number
1180 */
1181
1182 struct collect_highseqnum_db_state {
1183 struct tevent_context *ev;
1184 struct ctdb_client_context *client;
1185 uint32_t *pnn_list;
1186 int count;
1187 uint32_t *caps;
1188 uint32_t *ban_credits;
1189 uint32_t db_id;
1190 struct recdb_context *recdb;
1191 uint32_t max_pnn;
1192 };
1193
1194 static void collect_highseqnum_db_seqnum_done(struct tevent_req *subreq);
1195 static void collect_highseqnum_db_pulldb_done(struct tevent_req *subreq);
1196
1197 static struct tevent_req *collect_highseqnum_db_send(
1198 TALLOC_CTX *mem_ctx,
1199 struct tevent_context *ev,
1200 struct ctdb_client_context *client,
1201 uint32_t *pnn_list, int count, uint32_t *caps,
1202 uint32_t *ban_credits, uint32_t db_id,
1203 struct recdb_context *recdb)
1204 {
1205 struct tevent_req *req, *subreq;
1206 struct collect_highseqnum_db_state *state;
1207 struct ctdb_req_control request;
1208
1209 req = tevent_req_create(mem_ctx, &state,
1210 struct collect_highseqnum_db_state);
1211 if (req == NULL) {
1212 return NULL;
1213 }
1214
1215 state->ev = ev;
1216 state->client = client;
1217 state->pnn_list = pnn_list;
1218 state->count = count;
1219 state->caps = caps;
1220 state->ban_credits = ban_credits;
1221 state->db_id = db_id;
1222 state->recdb = recdb;
1223
1224 ctdb_req_control_get_db_seqnum(&request, db_id);
1225 subreq = ctdb_client_control_multi_send(mem_ctx, ev, client,
1226 state->pnn_list, state->count,
1227 TIMEOUT(), &request);
1228 if (tevent_req_nomem(subreq, req)) {
1229 return tevent_req_post(req, ev);
1230 }
1231 tevent_req_set_callback(subreq, collect_highseqnum_db_seqnum_done,
1232 req);
1233
1234 return req;
1235 }
1236
1237 static void collect_highseqnum_db_seqnum_done(struct tevent_req *subreq)
1238 {
1239 struct tevent_req *req = tevent_req_callback_data(
1240 subreq, struct tevent_req);
1241 struct collect_highseqnum_db_state *state = tevent_req_data(
1242 req, struct collect_highseqnum_db_state);
1243 struct ctdb_reply_control **reply;
1244 int *err_list;
1245 bool status;
1246 int ret, i;
1247 uint64_t seqnum, max_seqnum;
1248
1249 status = ctdb_client_control_multi_recv(subreq, &ret, state,
1250 &err_list, &reply);
1251 TALLOC_FREE(subreq);
1252 if (! status) {
1253 int ret2;
1254 uint32_t pnn;
1255
1256 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1257 state->count, err_list,
1258 &pnn);
1259 if (ret2 != 0) {
1260 D_ERR("control GET_DB_SEQNUM failed for db %s"
1261 " on node %u, ret=%d\n",
1262 recdb_name(state->recdb), pnn, ret2);
1263 } else {
1264 D_ERR("control GET_DB_SEQNUM failed for db %s,"
1265 " ret=%d\n",
1266 recdb_name(state->recdb), ret);
1267 }
1268 tevent_req_error(req, ret);
1269 return;
1270 }
1271
1272 max_seqnum = 0;
1273 state->max_pnn = state->pnn_list[0];
1274 for (i=0; i<state->count; i++) {
1275 ret = ctdb_reply_control_get_db_seqnum(reply[i], &seqnum);
1276 if (ret != 0) {
1277 tevent_req_error(req, EPROTO);
1278 return;
1279 }
1280
1281 if (max_seqnum < seqnum) {
1282 max_seqnum = seqnum;
1283 state->max_pnn = state->pnn_list[i];
1284 }
1285 }
1286
1287 talloc_free(reply);
1288
1289 D_INFO("Pull persistent db %s from node %d with seqnum 0x%"PRIx64"\n",
1290 recdb_name(state->recdb), state->max_pnn, max_seqnum);
1291
1292 subreq = pull_database_send(state, state->ev, state->client,
1293 state->max_pnn,
1294 state->caps[state->max_pnn],
1295 state->recdb);
1296 if (tevent_req_nomem(subreq, req)) {
1297 return;
1298 }
1299 tevent_req_set_callback(subreq, collect_highseqnum_db_pulldb_done,
1300 req);
1301 }
1302
1303 static void collect_highseqnum_db_pulldb_done(struct tevent_req *subreq)
1304 {
1305 struct tevent_req *req = tevent_req_callback_data(
1306 subreq, struct tevent_req);
1307 struct collect_highseqnum_db_state *state = tevent_req_data(
1308 req, struct collect_highseqnum_db_state);
1309 int ret;
1310 bool status;
1311
1312 status = pull_database_recv(subreq, &ret);
1313 TALLOC_FREE(subreq);
1314 if (! status) {
1315 state->ban_credits[state->max_pnn] += 1;
1316 tevent_req_error(req, ret);
1317 return;
1318 }
1319
1320 tevent_req_done(req);
1321 }
1322
1323 static bool collect_highseqnum_db_recv(struct tevent_req *req, int *perr)
1324 {
1325 return generic_recv(req, perr);
1326 }
1327
1328 /*
1329 * Collect all databases
1330 */
1331
1332 struct collect_all_db_state {
1333 struct tevent_context *ev;
1334 struct ctdb_client_context *client;
1335 uint32_t *pnn_list;
1336 int count;
1337 uint32_t *caps;
1338 uint32_t *ban_credits;
1339 uint32_t db_id;
1340 struct recdb_context *recdb;
1341 struct ctdb_pulldb pulldb;
1342 int index;
1343 };
1344
1345 static void collect_all_db_pulldb_done(struct tevent_req *subreq);
1346
1347 static struct tevent_req *collect_all_db_send(
1348 TALLOC_CTX *mem_ctx,
1349 struct tevent_context *ev,
1350 struct ctdb_client_context *client,
1351 uint32_t *pnn_list, int count, uint32_t *caps,
1352 uint32_t *ban_credits, uint32_t db_id,
1353 struct recdb_context *recdb)
1354 {
1355 struct tevent_req *req, *subreq;
1356 struct collect_all_db_state *state;
1357 uint32_t pnn;
1358
1359 req = tevent_req_create(mem_ctx, &state,
1360 struct collect_all_db_state);
1361 if (req == NULL) {
1362 return NULL;
1363 }
1364
1365 state->ev = ev;
1366 state->client = client;
1367 state->pnn_list = pnn_list;
1368 state->count = count;
1369 state->caps = caps;
1370 state->ban_credits = ban_credits;
1371 state->db_id = db_id;
1372 state->recdb = recdb;
1373 state->index = 0;
1374
1375 pnn = state->pnn_list[state->index];
1376
1377 subreq = pull_database_send(state, ev, client, pnn, caps[pnn], recdb);
1378 if (tevent_req_nomem(subreq, req)) {
1379 return tevent_req_post(req, ev);
1380 }
1381 tevent_req_set_callback(subreq, collect_all_db_pulldb_done, req);
1382
1383 return req;
1384 }
1385
1386 static void collect_all_db_pulldb_done(struct tevent_req *subreq)
1387 {
1388 struct tevent_req *req = tevent_req_callback_data(
1389 subreq, struct tevent_req);
1390 struct collect_all_db_state *state = tevent_req_data(
1391 req, struct collect_all_db_state);
1392 uint32_t pnn;
1393 int ret;
1394 bool status;
1395
1396 status = pull_database_recv(subreq, &ret);
1397 TALLOC_FREE(subreq);
1398 if (! status) {
1399 pnn = state->pnn_list[state->index];
1400 state->ban_credits[pnn] += 1;
1401 tevent_req_error(req, ret);
1402 return;
1403 }
1404
1405 state->index += 1;
1406 if (state->index == state->count) {
1407 tevent_req_done(req);
1408 return;
1409 }
1410
1411 pnn = state->pnn_list[state->index];
1412 subreq = pull_database_send(state, state->ev, state->client,
1413 pnn, state->caps[pnn], state->recdb);
1414 if (tevent_req_nomem(subreq, req)) {
1415 return;
1416 }
1417 tevent_req_set_callback(subreq, collect_all_db_pulldb_done, req);
1418 }
1419
1420 static bool collect_all_db_recv(struct tevent_req *req, int *perr)
1421 {
1422 return generic_recv(req, perr);
1423 }
1424
1425
1426 /**
1427 * For each database do the following:
1428 * - Get DB name
1429 * - Get DB path
1430 * - Freeze database on all nodes
1431 * - Start transaction on all nodes
1432 * - Collect database from all nodes
1433 * - Wipe database on all nodes
1434 * - Push database to all nodes
1435 * - Commit transaction on all nodes
1436 * - Thaw database on all nodes
1437 */
1438
1439 struct recover_db_state {
1440 struct tevent_context *ev;
1441 struct ctdb_client_context *client;
1442 struct ctdb_tunable_list *tun_list;
1443 uint32_t *pnn_list;
1444 int count;
1445 uint32_t *caps;
1446 uint32_t *ban_credits;
1447 uint32_t db_id;
1448 uint8_t db_flags;
1449
1450 uint32_t destnode;
1451 struct ctdb_transdb transdb;
1452
1453 const char *db_name, *db_path;
1454 struct recdb_context *recdb;
1455 };
1456
1457 static void recover_db_name_done(struct tevent_req *subreq);
1458 static void recover_db_path_done(struct tevent_req *subreq);
1459 static void recover_db_freeze_done(struct tevent_req *subreq);
1460 static void recover_db_transaction_started(struct tevent_req *subreq);
1461 static void recover_db_collect_done(struct tevent_req *subreq);
1462 static void recover_db_wipedb_done(struct tevent_req *subreq);
1463 static void recover_db_pushdb_done(struct tevent_req *subreq);
1464 static void recover_db_transaction_committed(struct tevent_req *subreq);
1465 static void recover_db_thaw_done(struct tevent_req *subreq);
1466
1467 static struct tevent_req *recover_db_send(TALLOC_CTX *mem_ctx,
1468 struct tevent_context *ev,
1469 struct ctdb_client_context *client,
1470 struct ctdb_tunable_list *tun_list,
1471 uint32_t *pnn_list, int count,
1472 uint32_t *caps,
1473 uint32_t *ban_credits,
1474 uint32_t generation,
1475 uint32_t db_id, uint8_t db_flags)
1476 {
1477 struct tevent_req *req, *subreq;
1478 struct recover_db_state *state;
1479 struct ctdb_req_control request;
1480
1481 req = tevent_req_create(mem_ctx, &state, struct recover_db_state);
1482 if (req == NULL) {
1483 return NULL;
1484 }
1485
1486 state->ev = ev;
1487 state->client = client;
1488 state->tun_list = tun_list;
1489 state->pnn_list = pnn_list;
1490 state->count = count;
1491 state->caps = caps;
1492 state->ban_credits = ban_credits;
1493 state->db_id = db_id;
1494 state->db_flags = db_flags;
1495
1496 state->destnode = ctdb_client_pnn(client);
1497 state->transdb.db_id = db_id;
1498 state->transdb.tid = generation;
1499
1500 ctdb_req_control_get_dbname(&request, db_id);
1501 subreq = ctdb_client_control_send(state, ev, client, state->destnode,
1502 TIMEOUT(), &request);
1503 if (tevent_req_nomem(subreq, req)) {
1504 return tevent_req_post(req, ev);
1505 }
1506 tevent_req_set_callback(subreq, recover_db_name_done, req);
1507
1508 return req;
1509 }
1510
1511 static void recover_db_name_done(struct tevent_req *subreq)
1512 {
1513 struct tevent_req *req = tevent_req_callback_data(
1514 subreq, struct tevent_req);
1515 struct recover_db_state *state = tevent_req_data(
1516 req, struct recover_db_state);
1517 struct ctdb_reply_control *reply;
1518 struct ctdb_req_control request;
1519 int ret;
1520 bool status;
1521
1522 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
1523 TALLOC_FREE(subreq);
1524 if (! status) {
1525 D_ERR("control GET_DBNAME failed for db=0x%x, ret=%d\n",
1526 state->db_id, ret);
1527 tevent_req_error(req, ret);
1528 return;
1529 }
1530
1531 ret = ctdb_reply_control_get_dbname(reply, state, &state->db_name);
1532 if (ret != 0) {
1533 D_ERR("control GET_DBNAME failed for db=0x%x, ret=%d\n",
1534 state->db_id, ret);
1535 tevent_req_error(req, EPROTO);
1536 return;
1537 }
1538
1539 talloc_free(reply);
1540
1541 ctdb_req_control_getdbpath(&request, state->db_id);
1542 subreq = ctdb_client_control_send(state, state->ev, state->client,
1543 state->destnode, TIMEOUT(),
1544 &request);
1545 if (tevent_req_nomem(subreq, req)) {
1546 return;
1547 }
1548 tevent_req_set_callback(subreq, recover_db_path_done, req);
1549 }
1550
1551 static void recover_db_path_done(struct tevent_req *subreq)
1552 {
1553 struct tevent_req *req = tevent_req_callback_data(
1554 subreq, struct tevent_req);
1555 struct recover_db_state *state = tevent_req_data(
1556 req, struct recover_db_state);
1557 struct ctdb_reply_control *reply;
1558 struct ctdb_req_control request;
1559 int ret;
1560 bool status;
1561
1562 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
1563 TALLOC_FREE(subreq);
1564 if (! status) {
1565 D_ERR("control GETDBPATH failed for db %s, ret=%d\n",
1566 state->db_name, ret);
1567 tevent_req_error(req, ret);
1568 return;
1569 }
1570
1571 ret = ctdb_reply_control_getdbpath(reply, state, &state->db_path);
1572 if (ret != 0) {
1573 D_ERR("control GETDBPATH failed for db %s, ret=%d\n",
1574 state->db_name, ret);
1575 tevent_req_error(req, EPROTO);
1576 return;
1577 }
1578
1579 talloc_free(reply);
1580
1581 ctdb_req_control_db_freeze(&request, state->db_id);
1582 subreq = ctdb_client_control_multi_send(state, state->ev,
1583 state->client,
1584 state->pnn_list, state->count,
1585 TIMEOUT(), &request);
1586 if (tevent_req_nomem(subreq, req)) {
1587 return;
1588 }
1589 tevent_req_set_callback(subreq, recover_db_freeze_done, req);
1590 }
1591
1592 static void recover_db_freeze_done(struct tevent_req *subreq)
1593 {
1594 struct tevent_req *req = tevent_req_callback_data(
1595 subreq, struct tevent_req);
1596 struct recover_db_state *state = tevent_req_data(
1597 req, struct recover_db_state);
1598 struct ctdb_req_control request;
1599 int *err_list;
1600 int ret;
1601 bool status;
1602
1603 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1604 NULL);
1605 TALLOC_FREE(subreq);
1606 if (! status) {
1607 int ret2;
1608 uint32_t pnn;
1609
1610 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1611 state->count, err_list,
1612 &pnn);
1613 if (ret2 != 0) {
1614 D_ERR("control FREEZE_DB failed for db %s"
1615 " on node %u, ret=%d\n",
1616 state->db_name, pnn, ret2);
1617 state->ban_credits[pnn] += 1;
1618 } else {
1619 D_ERR("control FREEZE_DB failed for db %s, ret=%d\n",
1620 state->db_name, ret);
1621 }
1622 tevent_req_error(req, ret);
1623 return;
1624 }
1625
1626 ctdb_req_control_db_transaction_start(&request, &state->transdb);
1627 subreq = ctdb_client_control_multi_send(state, state->ev,
1628 state->client,
1629 state->pnn_list, state->count,
1630 TIMEOUT(), &request);
1631 if (tevent_req_nomem(subreq, req)) {
1632 return;
1633 }
1634 tevent_req_set_callback(subreq, recover_db_transaction_started, req);
1635 }
1636
1637 static void recover_db_transaction_started(struct tevent_req *subreq)
1638 {
1639 struct tevent_req *req = tevent_req_callback_data(
1640 subreq, struct tevent_req);
1641 struct recover_db_state *state = tevent_req_data(
1642 req, struct recover_db_state);
1643 int *err_list;
1644 int ret;
1645 bool status;
1646
1647 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1648 NULL);
1649 TALLOC_FREE(subreq);
1650 if (! status) {
1651 int ret2;
1652 uint32_t pnn;
1653
1654 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1655 state->count,
1656 err_list, &pnn);
1657 if (ret2 != 0) {
1658 D_ERR("control TRANSACTION_DB failed for db=%s"
1659 " on node %u, ret=%d\n",
1660 state->db_name, pnn, ret2);
1661 } else {
1662 D_ERR("control TRANSACTION_DB failed for db=%s,"
1663 " ret=%d\n", state->db_name, ret);
1664 }
1665 tevent_req_error(req, ret);
1666 return;
1667 }
1668
1669 state->recdb = recdb_create(state, state->db_id, state->db_name,
1670 state->db_path,
1671 state->tun_list->database_hash_size,
1672 state->db_flags & CTDB_DB_FLAGS_PERSISTENT);
1673 if (tevent_req_nomem(state->recdb, req)) {
1674 return;
1675 }
1676
1677 if ((state->db_flags & CTDB_DB_FLAGS_PERSISTENT) ||
1678 (state->db_flags & CTDB_DB_FLAGS_REPLICATED)) {
1679 subreq = collect_highseqnum_db_send(
1680 state, state->ev, state->client,
1681 state->pnn_list, state->count, state->caps,
1682 state->ban_credits, state->db_id,
1683 state->recdb);
1684 } else {
1685 subreq = collect_all_db_send(
1686 state, state->ev, state->client,
1687 state->pnn_list, state->count, state->caps,
1688 state->ban_credits, state->db_id,
1689 state->recdb);
1690 }
1691 if (tevent_req_nomem(subreq, req)) {
1692 return;
1693 }
1694 tevent_req_set_callback(subreq, recover_db_collect_done, req);
1695 }
1696
1697 static void recover_db_collect_done(struct tevent_req *subreq)
1698 {
1699 struct tevent_req *req = tevent_req_callback_data(
1700 subreq, struct tevent_req);
1701 struct recover_db_state *state = tevent_req_data(
1702 req, struct recover_db_state);
1703 struct ctdb_req_control request;
1704 int ret;
1705 bool status;
1706
1707 if ((state->db_flags & CTDB_DB_FLAGS_PERSISTENT) ||
1708 (state->db_flags & CTDB_DB_FLAGS_REPLICATED)) {
1709 status = collect_highseqnum_db_recv(subreq, &ret);
1710 } else {
1711 status = collect_all_db_recv(subreq, &ret);
1712 }
1713 TALLOC_FREE(subreq);
1714 if (! status) {
1715 tevent_req_error(req, ret);
1716 return;
1717 }
1718
1719 ctdb_req_control_wipe_database(&request, &state->transdb);
1720 subreq = ctdb_client_control_multi_send(state, state->ev,
1721 state->client,
1722 state->pnn_list, state->count,
1723 TIMEOUT(), &request);
1724 if (tevent_req_nomem(subreq, req)) {
1725 return;
1726 }
1727 tevent_req_set_callback(subreq, recover_db_wipedb_done, req);
1728 }
1729
1730 static void recover_db_wipedb_done(struct tevent_req *subreq)
1731 {
1732 struct tevent_req *req = tevent_req_callback_data(
1733 subreq, struct tevent_req);
1734 struct recover_db_state *state = tevent_req_data(
1735 req, struct recover_db_state);
1736 int *err_list;
1737 int ret;
1738 bool status;
1739
1740 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1741 NULL);
1742 TALLOC_FREE(subreq);
1743 if (! status) {
1744 int ret2;
1745 uint32_t pnn;
1746
1747 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1748 state->count,
1749 err_list, &pnn);
1750 if (ret2 != 0) {
1751 D_ERR("control WIPEDB failed for db %s on node %u,"
1752 " ret=%d\n", state->db_name, pnn, ret2);
1753 } else {
1754 D_ERR("control WIPEDB failed for db %s, ret=%d\n",
1755 state->db_name, ret);
1756 }
1757 tevent_req_error(req, ret);
1758 return;
1759 }
1760
1761 subreq = push_database_send(state, state->ev, state->client,
1762 state->pnn_list, state->count,
1763 state->caps, state->tun_list,
1764 state->recdb);
1765 if (tevent_req_nomem(subreq, req)) {
1766 return;
1767 }
1768 tevent_req_set_callback(subreq, recover_db_pushdb_done, req);
1769 }
1770
1771 static void recover_db_pushdb_done(struct tevent_req *subreq)
1772 {
1773 struct tevent_req *req = tevent_req_callback_data(
1774 subreq, struct tevent_req);
1775 struct recover_db_state *state = tevent_req_data(
1776 req, struct recover_db_state);
1777 struct ctdb_req_control request;
1778 int ret;
1779 bool status;
1780
1781 status = push_database_recv(subreq, &ret);
1782 TALLOC_FREE(subreq);
1783 if (! status) {
1784 tevent_req_error(req, ret);
1785 return;
1786 }
1787
1788 TALLOC_FREE(state->recdb);
1789
1790 ctdb_req_control_db_transaction_commit(&request, &state->transdb);
1791 subreq = ctdb_client_control_multi_send(state, state->ev,
1792 state->client,
1793 state->pnn_list, state->count,
1794 TIMEOUT(), &request);
1795 if (tevent_req_nomem(subreq, req)) {
1796 return;
1797 }
1798 tevent_req_set_callback(subreq, recover_db_transaction_committed, req);
1799 }
1800
1801 static void recover_db_transaction_committed(struct tevent_req *subreq)
1802 {
1803 struct tevent_req *req = tevent_req_callback_data(
1804 subreq, struct tevent_req);
1805 struct recover_db_state *state = tevent_req_data(
1806 req, struct recover_db_state);
1807 struct ctdb_req_control request;
1808 int *err_list;
1809 int ret;
1810 bool status;
1811
1812 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1813 NULL);
1814 TALLOC_FREE(subreq);
1815 if (! status) {
1816 int ret2;
1817 uint32_t pnn;
1818
1819 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1820 state->count,
1821 err_list, &pnn);
1822 if (ret2 != 0) {
1823 D_ERR("control DB_TRANSACTION_COMMIT failed for db %s"
1824 " on node %u, ret=%d\n",
1825 state->db_name, pnn, ret2);
1826 } else {
1827 D_ERR("control DB_TRANSACTION_COMMIT failed for db %s,"
1828 " ret=%d\n", state->db_name, ret);
1829 }
1830 tevent_req_error(req, ret);
1831 return;
1832 }
1833
1834 ctdb_req_control_db_thaw(&request, state->db_id);
1835 subreq = ctdb_client_control_multi_send(state, state->ev,
1836 state->client,
1837 state->pnn_list, state->count,
1838 TIMEOUT(), &request);
1839 if (tevent_req_nomem(subreq, req)) {
1840 return;
1841 }
1842 tevent_req_set_callback(subreq, recover_db_thaw_done, req);
1843 }
1844
1845 static void recover_db_thaw_done(struct tevent_req *subreq)
1846 {
1847 struct tevent_req *req = tevent_req_callback_data(
1848 subreq, struct tevent_req);
1849 struct recover_db_state *state = tevent_req_data(
1850 req, struct recover_db_state);
1851 int *err_list;
1852 int ret;
1853 bool status;
1854
1855 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1856 NULL);
1857 TALLOC_FREE(subreq);
1858 if (! status) {
1859 int ret2;
1860 uint32_t pnn;
1861
1862 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1863 state->count,
1864 err_list, &pnn);
1865 if (ret2 != 0) {
1866 D_ERR("control DB_THAW failed for db %s on node %u,"
1867 " ret=%d\n", state->db_name, pnn, ret2);
1868 } else {
1869 D_ERR("control DB_THAW failed for db %s, ret=%d\n",
1870 state->db_name, ret);
1871 }
1872 tevent_req_error(req, ret);
1873 return;
1874 }
1875
1876 tevent_req_done(req);
1877 }
1878
1879 static bool recover_db_recv(struct tevent_req *req)
1880 {
1881 return generic_recv(req, NULL);
1882 }
1883
1884
1885 /*
1886 * Start database recovery for each database
1887 *
1888 * Try to recover each database 5 times before failing recovery.
1889 */
1890
1891 struct db_recovery_state {
1892 struct tevent_context *ev;
1893 struct ctdb_dbid_map *dbmap;
1894 int num_replies;
1895 int num_failed;
1896 };
1897
1898 struct db_recovery_one_state {
1899 struct tevent_req *req;
1900 struct ctdb_client_context *client;
1901 struct ctdb_dbid_map *dbmap;
1902 struct ctdb_tunable_list *tun_list;
1903 uint32_t *pnn_list;
1904 int count;
1905 uint32_t *caps;
1906 uint32_t *ban_credits;
1907 uint32_t generation;
1908 uint32_t db_id;
1909 uint8_t db_flags;
1910 int num_fails;
1911 };
1912
1913 static void db_recovery_one_done(struct tevent_req *subreq);
1914
1915 static struct tevent_req *db_recovery_send(TALLOC_CTX *mem_ctx,
1916 struct tevent_context *ev,
1917 struct ctdb_client_context *client,
1918 struct ctdb_dbid_map *dbmap,
1919 struct ctdb_tunable_list *tun_list,
1920 uint32_t *pnn_list, int count,
1921 uint32_t *caps,
1922 uint32_t *ban_credits,
1923 uint32_t generation)
1924 {
1925 struct tevent_req *req, *subreq;
1926 struct db_recovery_state *state;
1927 int i;
1928
1929 req = tevent_req_create(mem_ctx, &state, struct db_recovery_state);
1930 if (req == NULL) {
1931 return NULL;
1932 }
1933
1934 state->ev = ev;
1935 state->dbmap = dbmap;
1936 state->num_replies = 0;
1937 state->num_failed = 0;
1938
1939 if (dbmap->num == 0) {
1940 tevent_req_done(req);
1941 return tevent_req_post(req, ev);
1942 }
1943
1944 for (i=0; i<dbmap->num; i++) {
1945 struct db_recovery_one_state *substate;
1946
1947 substate = talloc_zero(state, struct db_recovery_one_state);
1948 if (tevent_req_nomem(substate, req)) {
1949 return tevent_req_post(req, ev);
1950 }
1951
1952 substate->req = req;
1953 substate->client = client;
1954 substate->dbmap = dbmap;
1955 substate->tun_list = tun_list;
1956 substate->pnn_list = pnn_list;
1957 substate->count = count;
1958 substate->caps = caps;
1959 substate->ban_credits = ban_credits;
1960 substate->generation = generation;
1961 substate->db_id = dbmap->dbs[i].db_id;
1962 substate->db_flags = dbmap->dbs[i].flags;
1963
1964 subreq = recover_db_send(state, ev, client, tun_list,
1965 pnn_list, count, caps, ban_credits,
1966 generation, substate->db_id,
1967 substate->db_flags);
1968 if (tevent_req_nomem(subreq, req)) {
1969 return tevent_req_post(req, ev);
1970 }
1971 tevent_req_set_callback(subreq, db_recovery_one_done,
1972 substate);
1973 D_NOTICE("recover database 0x%08x\n", substate->db_id);
1974 }
1975
1976 return req;
1977 }
1978
1979 static void db_recovery_one_done(struct tevent_req *subreq)
1980 {
1981 struct db_recovery_one_state *substate = tevent_req_callback_data(
1982 subreq, struct db_recovery_one_state);
1983 struct tevent_req *req = substate->req;
1984 struct db_recovery_state *state = tevent_req_data(
1985 req, struct db_recovery_state);
1986 bool status;
1987
1988 status = recover_db_recv(subreq);
1989 TALLOC_FREE(subreq);
1990
1991 if (status) {
1992 talloc_free(substate);
1993 goto done;
1994 }
1995
1996 substate->num_fails += 1;
1997 if (substate->num_fails < NUM_RETRIES) {
1998 subreq = recover_db_send(state, state->ev, substate->client,
1999 substate->tun_list,
2000 substate->pnn_list, substate->count,
2001 substate->caps, substate->ban_credits,
2002 substate->generation, substate->db_id,
2003 substate->db_flags);
2004 if (tevent_req_nomem(subreq, req)) {
2005 goto failed;
2006 }
2007 tevent_req_set_callback(subreq, db_recovery_one_done, substate);
2008 D_NOTICE("recover database 0x%08x, attempt %d\n",
2009 substate->db_id, substate->num_fails+1);
2010 return;
2011 }
2012
2013 failed:
2014 state->num_failed += 1;
2015
2016 done:
2017 state->num_replies += 1;
2018
2019 if (state->num_replies == state->dbmap->num) {
2020 tevent_req_done(req);
2021 }
2022 }
2023
2024 static bool db_recovery_recv(struct tevent_req *req, int *count)
2025 {
2026 struct db_recovery_state *state = tevent_req_data(
2027 req, struct db_recovery_state);
2028 int err;
2029
2030 if (tevent_req_is_unix_error(req, &err)) {
2031 *count = 0;
2032 return false;
2033 }
2034
2035 *count = state->num_replies - state->num_failed;
2036
2037 if (state->num_failed > 0) {
2038 return false;
2039 }
2040
2041 return true;
2042 }
2043
2044
2045 /*
2046 * Run the parallel database recovery
2047 *
2048 * - Get tunables
2049 * - Get nodemap
2050 * - Get vnnmap
2051 * - Get capabilities from all nodes
2052 * - Get dbmap
2053 * - Set RECOVERY_ACTIVE
2054 * - Send START_RECOVERY
2055 * - Update vnnmap on all nodes
2056 * - Run database recovery
2057 * - Set RECOVERY_NORMAL
2058 * - Send END_RECOVERY
2059 */
2060
2061 struct recovery_state {
2062 struct tevent_context *ev;
2063 struct ctdb_client_context *client;
2064 uint32_t generation;
2065 uint32_t *pnn_list;
2066 int count;
2067 uint32_t destnode;
2068 struct ctdb_node_map *nodemap;
2069 uint32_t *caps;
2070 uint32_t *ban_credits;
2071 struct ctdb_tunable_list *tun_list;
2072 struct ctdb_vnn_map *vnnmap;
2073 struct ctdb_dbid_map *dbmap;
2074 };
2075
2076 static void recovery_tunables_done(struct tevent_req *subreq);
2077 static void recovery_nodemap_done(struct tevent_req *subreq);
2078 static void recovery_vnnmap_done(struct tevent_req *subreq);
2079 static void recovery_capabilities_done(struct tevent_req *subreq);
2080 static void recovery_dbmap_done(struct tevent_req *subreq);
2081 static void recovery_active_done(struct tevent_req *subreq);
2082 static void recovery_start_recovery_done(struct tevent_req *subreq);
2083 static void recovery_vnnmap_update_done(struct tevent_req *subreq);
2084 static void recovery_db_recovery_done(struct tevent_req *subreq);
2085 static void recovery_failed_done(struct tevent_req *subreq);
2086 static void recovery_normal_done(struct tevent_req *subreq);
2087 static void recovery_end_recovery_done(struct tevent_req *subreq);
2088
2089 static struct tevent_req *recovery_send(TALLOC_CTX *mem_ctx,
2090 struct tevent_context *ev,
2091 struct ctdb_client_context *client,
2092 uint32_t generation)
2093 {
2094 struct tevent_req *req, *subreq;
2095 struct recovery_state *state;
2096 struct ctdb_req_control request;
2097
2098 req = tevent_req_create(mem_ctx, &state, struct recovery_state);
2099 if (req == NULL) {
2100 return NULL;
2101 }
2102
2103 state->ev = ev;
2104 state->client = client;
2105 state->generation = generation;
2106 state->destnode = ctdb_client_pnn(client);
2107
2108 ctdb_req_control_get_all_tunables(&request);
2109 subreq = ctdb_client_control_send(state, state->ev, state->client,
2110 state->destnode, TIMEOUT(),
2111 &request);
2112 if (tevent_req_nomem(subreq, req)) {
2113 return tevent_req_post(req, ev);
2114 }
2115 tevent_req_set_callback(subreq, recovery_tunables_done, req);
2116
2117 return req;
2118 }
2119
2120 static void recovery_tunables_done(struct tevent_req *subreq)
2121 {
2122 struct tevent_req *req = tevent_req_callback_data(
2123 subreq, struct tevent_req);
2124 struct recovery_state *state = tevent_req_data(
2125 req, struct recovery_state);
2126 struct ctdb_reply_control *reply;
2127 struct ctdb_req_control request;
2128 int ret;
2129 bool status;
2130
2131 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2132 TALLOC_FREE(subreq);
2133 if (! status) {
2134 D_ERR("control GET_ALL_TUNABLES failed, ret=%d\n", ret);
2135 tevent_req_error(req, ret);
2136 return;
2137 }
2138
2139 ret = ctdb_reply_control_get_all_tunables(reply, state,
2140 &state->tun_list);
2141 if (ret != 0) {
2142 D_ERR("control GET_ALL_TUNABLES failed, ret=%d\n", ret);
2143 tevent_req_error(req, EPROTO);
2144 return;
2145 }
2146
2147 talloc_free(reply);
2148
2149 recover_timeout = state->tun_list->recover_timeout;
2150
2151 ctdb_req_control_get_nodemap(&request);
2152 subreq = ctdb_client_control_send(state, state->ev, state->client,
2153 state->destnode, TIMEOUT(),
2154 &request);
2155 if (tevent_req_nomem(subreq, req)) {
2156 return;
2157 }
2158 tevent_req_set_callback(subreq, recovery_nodemap_done, req);
2159 }
2160
2161 static void recovery_nodemap_done(struct tevent_req *subreq)
2162 {
2163 struct tevent_req *req = tevent_req_callback_data(
2164 subreq, struct tevent_req);
2165 struct recovery_state *state = tevent_req_data(
2166 req, struct recovery_state);
2167 struct ctdb_reply_control *reply;
2168 struct ctdb_req_control request;
2169 bool status;
2170 int ret;
2171
2172 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2173 TALLOC_FREE(subreq);
2174 if (! status) {
2175 D_ERR("control GET_NODEMAP failed to node %u, ret=%d\n",
2176 state->destnode, ret);
2177 tevent_req_error(req, ret);
2178 return;
2179 }
2180
2181 ret = ctdb_reply_control_get_nodemap(reply, state, &state->nodemap);
2182 if (ret != 0) {
2183 D_ERR("control GET_NODEMAP failed, ret=%d\n", ret);
2184 tevent_req_error(req, ret);
2185 return;
2186 }
2187
2188 state->count = list_of_active_nodes(state->nodemap, CTDB_UNKNOWN_PNN,
2189 state, &state->pnn_list);
2190 if (state->count <= 0) {
2191 tevent_req_error(req, ENOMEM);
2192 return;
2193 }
2194
2195 state->ban_credits = talloc_zero_array(state, uint32_t,
2196 state->nodemap->num);
2197 if (tevent_req_nomem(state->ban_credits, req)) {
2198 return;
2199 }
2200
2201 ctdb_req_control_getvnnmap(&request);
2202 subreq = ctdb_client_control_send(state, state->ev, state->client,
2203 state->destnode, TIMEOUT(),
2204 &request);
2205 if (tevent_req_nomem(subreq, req)) {
2206 return;
2207 }
2208 tevent_req_set_callback(subreq, recovery_vnnmap_done, req);
2209 }
2210
2211 static void recovery_vnnmap_done(struct tevent_req *subreq)
2212 {
2213 struct tevent_req *req = tevent_req_callback_data(
2214 subreq, struct tevent_req);
2215 struct recovery_state *state = tevent_req_data(
2216 req, struct recovery_state);
2217 struct ctdb_reply_control *reply;
2218 struct ctdb_req_control request;
2219 bool status;
2220 int ret;
2221
2222 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2223 TALLOC_FREE(subreq);
2224 if (! status) {
2225 D_ERR("control GETVNNMAP failed to node %u, ret=%d\n",
2226 state->destnode, ret);
2227 tevent_req_error(req, ret);
2228 return;
2229 }
2230
2231 ret = ctdb_reply_control_getvnnmap(reply, state, &state->vnnmap);
2232 if (ret != 0) {
2233 D_ERR("control GETVNNMAP failed, ret=%d\n", ret);
2234 tevent_req_error(req, ret);
2235 return;
2236 }
2237
2238 ctdb_req_control_get_capabilities(&request);
2239 subreq = ctdb_client_control_multi_send(state, state->ev,
2240 state->client,
2241 state->pnn_list, state->count,
2242 TIMEOUT(), &request);
2243 if (tevent_req_nomem(subreq, req)) {
2244 return;
2245 }
2246 tevent_req_set_callback(subreq, recovery_capabilities_done, req);
2247 }
2248
2249 static void recovery_capabilities_done(struct tevent_req *subreq)
2250 {
2251 struct tevent_req *req = tevent_req_callback_data(
2252 subreq, struct tevent_req);
2253 struct recovery_state *state = tevent_req_data(
2254 req, struct recovery_state);
2255 struct ctdb_reply_control **reply;
2256 struct ctdb_req_control request;
2257 int *err_list;
2258 int ret, i;
2259 bool status;
2260
2261 status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
2262 &reply);
2263 TALLOC_FREE(subreq);
2264 if (! status) {
2265 int ret2;
2266 uint32_t pnn;
2267
2268 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2269 state->count,
2270 err_list, &pnn);
2271 if (ret2 != 0) {
2272 D_ERR("control GET_CAPABILITIES failed on node %u,"
2273 " ret=%d\n", pnn, ret2);
2274 } else {
2275 D_ERR("control GET_CAPABILITIES failed, ret=%d\n",
2276 ret);
2277 }
2278 tevent_req_error(req, ret);
2279 return;
2280 }
2281
2282 /* Make the array size same as nodemap */
2283 state->caps = talloc_zero_array(state, uint32_t,
2284 state->nodemap->num);
2285 if (tevent_req_nomem(state->caps, req)) {
2286 return;
2287 }
2288
2289 for (i=0; i<state->count; i++) {
2290 uint32_t pnn;
2291
2292 pnn = state->pnn_list[i];
2293 ret = ctdb_reply_control_get_capabilities(reply[i],
2294 &state->caps[pnn]);
2295 if (ret != 0) {
2296 D_ERR("control GET_CAPABILITIES failed on node %u\n",
2297 pnn);
2298 tevent_req_error(req, EPROTO);
2299 return;
2300 }
2301 }
2302
2303 talloc_free(reply);
2304
2305 ctdb_req_control_get_dbmap(&request);
2306 subreq = ctdb_client_control_send(state, state->ev, state->client,
2307 state->destnode, TIMEOUT(),
2308 &request);
2309 if (tevent_req_nomem(subreq, req)) {
2310 return;
2311 }
2312 tevent_req_set_callback(subreq, recovery_dbmap_done, req);
2313 }
2314
2315 static void recovery_dbmap_done(struct tevent_req *subreq)
2316 {
2317 struct tevent_req *req = tevent_req_callback_data(
2318 subreq, struct tevent_req);
2319 struct recovery_state *state = tevent_req_data(
2320 req, struct recovery_state);
2321 struct ctdb_reply_control *reply;
2322 struct ctdb_req_control request;
2323 int ret;
2324 bool status;
2325
2326 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2327 TALLOC_FREE(subreq);
2328 if (! status) {
2329 D_ERR("control GET_DBMAP failed to node %u, ret=%d\n",
2330 state->destnode, ret);
2331 tevent_req_error(req, ret);
2332 return;
2333 }
2334
2335 ret = ctdb_reply_control_get_dbmap(reply, state, &state->dbmap);
2336 if (ret != 0) {
2337 D_ERR("control GET_DBMAP failed, ret=%d\n", ret);
2338 tevent_req_error(req, ret);
2339 return;
2340 }
2341
2342 ctdb_req_control_set_recmode(&request, CTDB_RECOVERY_ACTIVE);
2343 subreq = ctdb_client_control_multi_send(state, state->ev,
2344 state->client,
2345 state->pnn_list, state->count,
2346 TIMEOUT(), &request);
2347 if (tevent_req_nomem(subreq, req)) {
2348 return;
2349 }
2350 tevent_req_set_callback(subreq, recovery_active_done, req);
2351 }
2352
2353 static void recovery_active_done(struct tevent_req *subreq)
2354 {
2355 struct tevent_req *req = tevent_req_callback_data(
2356 subreq, struct tevent_req);
2357 struct recovery_state *state = tevent_req_data(
2358 req, struct recovery_state);
2359 struct ctdb_req_control request;
2360 struct ctdb_vnn_map *vnnmap;
2361 int *err_list;
2362 int ret, i;
2363 unsigned int count;
2364 bool status;
2365
2366 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
2367 NULL);
2368 TALLOC_FREE(subreq);
2369 if (! status) {
2370 int ret2;
2371 uint32_t pnn;
2372
2373 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2374 state->count,
2375 err_list, &pnn);
2376 if (ret2 != 0) {
2377 D_ERR("failed to set recovery mode ACTIVE on node %u,"
2378 " ret=%d\n", pnn, ret2);
2379 } else {
2380 D_ERR("failed to set recovery mode ACTIVE, ret=%d\n",
2381 ret);
2382 }
2383 tevent_req_error(req, ret);
2384 return;
2385 }
2386
2387 D_ERR("Set recovery mode to ACTIVE\n");
2388
2389 /* Calculate new VNNMAP */
2390 count = 0;
2391 for (i=0; i<state->nodemap->num; i++) {
2392 if (state->nodemap->node[i].flags & NODE_FLAGS_INACTIVE) {
2393 continue;
2394 }
2395 if (!(state->caps[i] & CTDB_CAP_LMASTER)) {
2396 continue;
2397 }
2398 count += 1;
2399 }
2400
2401 if (count == 0) {
2402 D_WARNING("No active lmasters found. Adding recmaster anyway\n");
2403 }
2404
2405 vnnmap = talloc_zero(state, struct ctdb_vnn_map);
2406 if (tevent_req_nomem(vnnmap, req)) {
2407 return;
2408 }
2409
2410 vnnmap->size = (count == 0 ? 1 : count);
2411 vnnmap->map = talloc_array(vnnmap, uint32_t, vnnmap->size);
2412 if (tevent_req_nomem(vnnmap->map, req)) {
2413 return;
2414 }
2415
2416 if (count == 0) {
2417 vnnmap->map[0] = state->destnode;
2418 } else {
2419 count = 0;
2420 for (i=0; i<state->nodemap->num; i++) {
2421 if (state->nodemap->node[i].flags &
2422 NODE_FLAGS_INACTIVE) {
2423 continue;
2424 }
2425 if (!(state->caps[i] & CTDB_CAP_LMASTER)) {
2426 continue;
2427 }
2428
2429 vnnmap->map[count] = state->nodemap->node[i].pnn;
2430 count += 1;
2431 }
2432 }
2433
2434 vnnmap->generation = state->generation;
2435
2436 talloc_free(state->vnnmap);
2437 state->vnnmap = vnnmap;
2438
2439 ctdb_req_control_start_recovery(&request);
2440 subreq = ctdb_client_control_multi_send(state, state->ev,
2441 state->client,
2442 state->pnn_list, state->count,
2443 TIMEOUT(), &request);
2444 if (tevent_req_nomem(subreq, req)) {
2445 return;
2446 }
2447 tevent_req_set_callback(subreq, recovery_start_recovery_done, req);
2448 }
2449
2450 static void recovery_start_recovery_done(struct tevent_req *subreq)
2451 {
2452 struct tevent_req *req = tevent_req_callback_data(
2453 subreq, struct tevent_req);
2454 struct recovery_state *state = tevent_req_data(
2455 req, struct recovery_state);
2456 struct ctdb_req_control request;
2457 int *err_list;
2458 int ret;
2459 bool status;
2460
2461 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
2462 NULL);
2463 TALLOC_FREE(subreq);
2464 if (! status) {
2465 int ret2;
2466 uint32_t pnn;
2467
2468 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2469 state->count,
2470 err_list, &pnn);
2471 if (ret2 != 0) {
2472 D_ERR("failed to run start_recovery event on node %u,"
2473 " ret=%d\n", pnn, ret2);
2474 } else {
2475 D_ERR("failed to run start_recovery event, ret=%d\n",
2476 ret);
2477 }
2478 tevent_req_error(req, ret);
2479 return;
2480 }
2481
2482 D_ERR("start_recovery event finished\n");
2483
2484 ctdb_req_control_setvnnmap(&request, state->vnnmap);
2485 subreq = ctdb_client_control_multi_send(state, state->ev,
2486 state->client,
2487 state->pnn_list, state->count,
2488 TIMEOUT(), &request);
2489 if (tevent_req_nomem(subreq, req)) {
2490 return;
2491 }
2492 tevent_req_set_callback(subreq, recovery_vnnmap_update_done, req);
2493 }
2494
2495 static void recovery_vnnmap_update_done(struct tevent_req *subreq)
2496 {
2497 struct tevent_req *req = tevent_req_callback_data(
2498 subreq, struct tevent_req);
2499 struct recovery_state *state = tevent_req_data(
2500 req, struct recovery_state);
2501 int *err_list;
2502 int ret;
2503 bool status;
2504
2505 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
2506 NULL);
2507 TALLOC_FREE(subreq);
2508 if (! status) {
2509 int ret2;
2510 uint32_t pnn;
2511
2512 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2513 state->count,
2514 err_list, &pnn);
2515 if (ret2 != 0) {
2516 D_ERR("failed to update VNNMAP on node %u, ret=%d\n",
2517 pnn, ret2);
2518 } else {
2519 D_ERR("failed to update VNNMAP, ret=%d\n", ret);
2520 }
2521 tevent_req_error(req, ret);
2522 return;
2523 }
2524
2525 D_NOTICE("updated VNNMAP\n");
2526
2527 subreq = db_recovery_send(state, state->ev, state->client,
2528 state->dbmap, state->tun_list,
2529 state->pnn_list, state->count,
2530 state->caps, state->ban_credits,
2531 state->vnnmap->generation);
2532 if (tevent_req_nomem(subreq, req)) {
2533 return;
2534 }
2535 tevent_req_set_callback(subreq, recovery_db_recovery_done, req);
2536 }
2537
2538 static void recovery_db_recovery_done(struct tevent_req *subreq)
2539 {
2540 struct tevent_req *req = tevent_req_callback_data(
2541 subreq, struct tevent_req);
2542 struct recovery_state *state = tevent_req_data(
2543 req, struct recovery_state);
2544 struct ctdb_req_control request;
2545 bool status;
2546 int count;
2547
2548 status = db_recovery_recv(subreq, &count);
2549 TALLOC_FREE(subreq);
2550
2551 D_ERR("%d of %d databases recovered\n", count, state->dbmap->num);
2552
2553 if (! status) {
2554 uint32_t max_pnn = CTDB_UNKNOWN_PNN, max_credits = 0;
2555 int i;
2556
2557 /* Bans are not enabled */
2558 if (state->tun_list->enable_bans == 0) {
2559 tevent_req_error(req, EIO);
2560 return;
2561 }
2562
2563 for (i=0; i<state->count; i++) {
2564 uint32_t pnn;
2565 pnn = state->pnn_list[i];
2566 if (state->ban_credits[pnn] > max_credits) {
2567 max_pnn = pnn;
2568 max_credits = state->ban_credits[pnn];
2569 }
2570 }
2571
2572 /* If pulling database fails multiple times */
2573 if (max_credits >= NUM_RETRIES) {
2574 struct ctdb_ban_state ban_state = {
2575 .pnn = max_pnn,
2576 .time = state->tun_list->recovery_ban_period,
2577 };
2578
2579 D_ERR("Banning node %u for %u seconds\n",
2580 ban_state.pnn,
2581 ban_state.time);
2582
2583 ctdb_req_control_set_ban_state(&request,
2584 &ban_state);
2585 subreq = ctdb_client_control_send(state,
2586 state->ev,
2587 state->client,
2588 ban_state.pnn,
2589 TIMEOUT(),
2590 &request);
2591 if (tevent_req_nomem(subreq, req)) {
2592 return;
2593 }
2594 tevent_req_set_callback(subreq,
2595 recovery_failed_done,
2596 req);
2597 } else {
2598 tevent_req_error(req, EIO);
2599 }
2600 return;
2601 }
2602
2603 ctdb_req_control_set_recmode(&request, CTDB_RECOVERY_NORMAL);
2604 subreq = ctdb_client_control_multi_send(state, state->ev,
2605 state->client,
2606 state->pnn_list, state->count,
2607 TIMEOUT(), &request);
2608 if (tevent_req_nomem(subreq, req)) {
2609 return;
2610 }
2611 tevent_req_set_callback(subreq, recovery_normal_done, req);
2612 }
2613
2614 static void recovery_failed_done(struct tevent_req *subreq)
2615 {
2616 struct tevent_req *req = tevent_req_callback_data(
2617 subreq, struct tevent_req);
2618 struct recovery_state *state = tevent_req_data(
2619 req, struct recovery_state);
2620 struct ctdb_reply_control *reply;
2621 int ret;
2622 bool status;
2623
2624 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2625 TALLOC_FREE(subreq);
2626 if (! status) {
2627 D_ERR("failed to ban node, ret=%d\n", ret);
2628 goto done;
2629 }
2630
2631 ret = ctdb_reply_control_set_ban_state(reply);
2632 if (ret != 0) {
2633 D_ERR("control SET_BAN_STATE failed, ret=%d\n", ret);
2634 }
2635
2636 done:
2637 tevent_req_error(req, EIO);
2638 }
2639
2640 static void recovery_normal_done(struct tevent_req *subreq)
2641 {
2642 struct tevent_req *req = tevent_req_callback_data(
2643 subreq, struct tevent_req);
2644 struct recovery_state *state = tevent_req_data(
2645 req, struct recovery_state);
2646 struct ctdb_req_control request;
2647 int *err_list;
2648 int ret;
2649 bool status;
2650
2651 status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
2652 NULL);
2653 TALLOC_FREE(subreq);
2654 if (! status) {
2655 int ret2;
2656 uint32_t pnn;
2657
2658 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2659 state->count,
2660 err_list, &pnn);
2661 if (ret2 != 0) {
2662 D_ERR("failed to set recovery mode NORMAL on node %u,"
2663 " ret=%d\n", pnn, ret2);
2664 } else {
2665 D_ERR("failed to set recovery mode NORMAL, ret=%d\n",
2666 ret);
2667 }
2668 tevent_req_error(req, ret);
2669 return;
2670 }
2671
2672 D_ERR("Set recovery mode to NORMAL\n");
2673
2674 ctdb_req_control_end_recovery(&request);
2675 subreq = ctdb_client_control_multi_send(state, state->ev,
2676 state->client,
2677 state->pnn_list, state->count,
2678 TIMEOUT(), &request);
2679 if (tevent_req_nomem(subreq, req)) {
2680 return;
2681 }
2682 tevent_req_set_callback(subreq, recovery_end_recovery_done, req);
2683 }
2684
2685 static void recovery_end_recovery_done(struct tevent_req *subreq)
2686 {
2687 struct tevent_req *req = tevent_req_callback_data(
2688 subreq, struct tevent_req);
2689 struct recovery_state *state = tevent_req_data(
2690 req, struct recovery_state);
2691 int *err_list;
2692 int ret;
2693 bool status;
2694
2695 status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
2696 NULL);
2697 TALLOC_FREE(subreq);
2698 if (! status) {
2699 int ret2;
2700 uint32_t pnn;
2701
2702 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2703 state->count,
2704 err_list, &pnn);
2705 if (ret2 != 0) {
2706 D_ERR("failed to run recovered event on node %u,"
2707 " ret=%d\n", pnn, ret2);
2708 } else {
2709 D_ERR("failed to run recovered event, ret=%d\n", ret);
2710 }
2711 tevent_req_error(req, ret);
2712 return;
2713 }
2714
2715 D_ERR("recovered event finished\n");
2716
2717 tevent_req_done(req);
2718 }
2719
2720 static void recovery_recv(struct tevent_req *req, int *perr)
2721 {
2722 generic_recv(req, perr);
2723 }
2724
2725 static void usage(const char *progname)
2726 {
2727 fprintf(stderr, "\nUsage: %s <output-fd> <ctdb-socket-path> <generation>\n",
2728 progname);
2729 }
2730
2731
2732 /*
2733 * Arguments - log fd, write fd, socket path, generation
2734 */
2735 int main(int argc, char *argv[])
2736 {
2737 int write_fd;
2738 const char *sockpath;
2739 TALLOC_CTX *mem_ctx;
2740 struct tevent_context *ev;
2741 struct ctdb_client_context *client;
2742 int ret;
2743 struct tevent_req *req;
2744 uint32_t generation;
2745
2746 if (argc != 4) {
2747 usage(argv[0]);
2748 exit(1);
2749 }
2750
2751 write_fd = atoi(argv[1]);
2752 sockpath = argv[2];
2753 generation = (uint32_t)strtoul(argv[3], NULL, 0);
2754
2755 mem_ctx = talloc_new(NULL);
2756 if (mem_ctx == NULL) {
2757 fprintf(stderr, "recovery: talloc_new() failed\n");
2758 goto failed;
2759 }
2760
2761 ret = logging_init(mem_ctx, NULL, NULL, "ctdb-recovery");
2762 if (ret != 0) {
2763 fprintf(stderr, "recovery: Unable to initialize logging\n");
2764 goto failed;
2765 }
2766
2767 ev = tevent_context_init(mem_ctx);
2768 if (ev == NULL) {
2769 D_ERR("tevent_context_init() failed\n");
2770 goto failed;
2771 }
2772
2773 ret = ctdb_client_init(mem_ctx, ev, sockpath, &client);
2774 if (ret != 0) {
2775 D_ERR("ctdb_client_init() failed, ret=%d\n", ret);
2776 goto failed;
2777 }
2778
2779 req = recovery_send(mem_ctx, ev, client, generation);
2780 if (req == NULL) {
2781 D_ERR("database_recover_send() failed\n");
2782 goto failed;
2783 }
2784
2785 if (! tevent_req_poll(req, ev)) {
2786 D_ERR("tevent_req_poll() failed\n");
2787 goto failed;
2788 }
2789
2790 recovery_recv(req, &ret);
2791 TALLOC_FREE(req);
2792 if (ret != 0) {
2793 D_ERR("database recovery failed, ret=%d\n", ret);
2794 goto failed;
2795 }
2796
2797 sys_write(write_fd, &ret, sizeof(ret));
2798 return 0;
2799
2800 failed:
2801 TALLOC_FREE(mem_ctx);
2802 return 1;
2803 }
Samba GIT mirror