?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Convert uint32/16/8 to _t for the last two include files in source3/include.
[Samba.git] / ctdb / server / ctdb_ltdb_server.c
blob174a460178453c1b7bd7e32bcea533ab4c2ace26
1 /*
2 ctdb ltdb code - server side
3
4 Copyright (C) Andrew Tridgell 2007
5
6 This program is free software; you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
10
11 This program is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
15
16 You should have received a copy of the GNU General Public License
17 along with this program; if not, see <http://www.gnu.org/licenses/>.
18 */
19
20 #include "includes.h"
21 #include "tdb.h"
22 #include "system/network.h"
23 #include "system/filesys.h"
24 #include "system/dir.h"
25 #include "system/time.h"
26 #include "../include/ctdb_private.h"
27 #include "../common/rb_tree.h"
28 #include "lib/tdb_wrap/tdb_wrap.h"
29 #include "lib/util/dlinklist.h"
30 #include <ctype.h>
31
32 #define PERSISTENT_HEALTH_TDB "persistent_health.tdb"
33
34 /**
35 * write a record to a normal database
36 *
37 * This is the server-variant of the ctdb_ltdb_store function.
38 * It contains logic to determine whether a record should be
39 * stored or deleted. It also sends SCHEDULE_FOR_DELETION
40 * controls to the local ctdb daemon if apporpriate.
41 */
42 static int ctdb_ltdb_store_server(struct ctdb_db_context *ctdb_db,
43 TDB_DATA key,
44 struct ctdb_ltdb_header *header,
45 TDB_DATA data)
46 {
47 struct ctdb_context *ctdb = ctdb_db->ctdb;
48 TDB_DATA rec;
49 int ret;
50 bool seqnum_suppressed = false;
51 bool keep = false;
52 bool schedule_for_deletion = false;
53 bool remove_from_delete_queue = false;
54 uint32_t lmaster;
55
56 if (ctdb->flags & CTDB_FLAG_TORTURE) {
57 struct ctdb_ltdb_header *h2;
58 rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
59 h2 = (struct ctdb_ltdb_header *)rec.dptr;
60 if (rec.dptr && rec.dsize >= sizeof(h2) && h2->rsn > header->rsn) {
61 DEBUG(DEBUG_CRIT,("RSN regression! %llu %llu\n",
62 (unsigned long long)h2->rsn, (unsigned long long)header->rsn));
63 }
64 if (rec.dptr) free(rec.dptr);
65 }
66
67 if (ctdb->vnn_map == NULL) {
68 /*
69 * Called from a client: always store the record
70 * Also don't call ctdb_lmaster since it uses the vnn_map!
71 */
72 keep = true;
73 goto store;
74 }
75
76 lmaster = ctdb_lmaster(ctdb_db->ctdb, &key);
77
78 /*
79 * If we migrate an empty record off to another node
80 * and the record has not been migrated with data,
81 * delete the record instead of storing the empty record.
82 */
83 if (data.dsize != 0) {
84 keep = true;
85 } else if (header->flags & CTDB_REC_RO_FLAGS) {
86 keep = true;
87 } else if (ctdb_db->persistent) {
88 keep = true;
89 } else if (header->flags & CTDB_REC_FLAG_AUTOMATIC) {
90 /*
91 * The record is not created by the client but
92 * automatically by the ctdb_ltdb_fetch logic that
93 * creates a record with an initial header in the
94 * ltdb before trying to migrate the record from
95 * the current lmaster. Keep it instead of trying
96 * to delete the non-existing record...
97 */
98 keep = true;
99 schedule_for_deletion = true;
100 } else if (header->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) {
101 keep = true;
102 } else if (ctdb_db->ctdb->pnn == lmaster) {
103 /*
104 * If we are lmaster, then we usually keep the record.
105 * But if we retrieve the dmaster role by a VACUUM_MIGRATE
106 * and the record is empty and has never been migrated
107 * with data, then we should delete it instead of storing it.
108 * This is part of the vacuuming process.
109 *
110 * The reason that we usually need to store even empty records
111 * on the lmaster is that a client operating directly on the
112 * lmaster (== dmaster) expects the local copy of the record to
113 * exist after successful ctdb migrate call. If the record does
114 * not exist, the client goes into a migrate loop and eventually
115 * fails. So storing the empty record makes sure that we do not
116 * need to change the client code.
117 */
118 if ((header->flags & CTDB_REC_FLAG_VACUUM_MIGRATED) &&
119 (ctdb_db->ctdb->pnn == header->dmaster)) {
120 keep = true;
121 schedule_for_deletion = true;
122 }
123 if (!(header->flags & CTDB_REC_FLAG_VACUUM_MIGRATED)) {
124 keep = true;
125 } else if (ctdb_db->ctdb->pnn != header->dmaster) {
126 keep = true;
127 }
128 } else if (ctdb_db->ctdb->pnn == header->dmaster) {
129 keep = true;
130 }
131
132 if (keep) {
133 if (!ctdb_db->persistent &&
134 (ctdb_db->ctdb->pnn == header->dmaster) &&
135 !(header->flags & CTDB_REC_RO_FLAGS))
136 {
137 header->rsn++;
138
139 if (data.dsize == 0) {
140 schedule_for_deletion = true;
141 }
142 }
143 remove_from_delete_queue = !schedule_for_deletion;
144 }
145
146 store:
147 /*
148 * The VACUUM_MIGRATED flag is only set temporarily for
149 * the above logic when the record was retrieved by a
150 * VACUUM_MIGRATE call and should not be stored in the
151 * database.
152 *
153 * The VACUUM_MIGRATE call is triggered by a vacuum fetch,
154 * and there are two cases in which the corresponding record
155 * is stored in the local database:
156 * 1. The record has been migrated with data in the past
157 * (the MIGRATED_WITH_DATA record flag is set).
158 * 2. The record has been filled with data again since it
159 * had been submitted in the VACUUM_FETCH message to the
160 * lmaster.
161 * For such records it is important to not store the
162 * VACUUM_MIGRATED flag in the database.
163 */
164 header->flags &= ~CTDB_REC_FLAG_VACUUM_MIGRATED;
165
166 /*
167 * Similarly, clear the AUTOMATIC flag which should not enter
168 * the local database copy since this would require client
169 * modifications to clear the flag when the client stores
170 * the record.
171 */
172 header->flags &= ~CTDB_REC_FLAG_AUTOMATIC;
173
174 rec.dsize = sizeof(*header) + data.dsize;
175 rec.dptr = talloc_size(ctdb, rec.dsize);
176 CTDB_NO_MEMORY(ctdb, rec.dptr);
177
178 memcpy(rec.dptr, header, sizeof(*header));
179 memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
180
181 /* Databases with seqnum updates enabled only get their seqnum
182 changes when/if we modify the data */
183 if (ctdb_db->seqnum_update != NULL) {
184 TDB_DATA old;
185 old = tdb_fetch(ctdb_db->ltdb->tdb, key);
186
187 if ( (old.dsize == rec.dsize)
188 && !memcmp(old.dptr+sizeof(struct ctdb_ltdb_header),
189 rec.dptr+sizeof(struct ctdb_ltdb_header),
190 rec.dsize-sizeof(struct ctdb_ltdb_header)) ) {
191 tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
192 seqnum_suppressed = true;
193 }
194 if (old.dptr) free(old.dptr);
195 }
196
197 DEBUG(DEBUG_DEBUG, (__location__ " db[%s]: %s record: hash[0x%08x]\n",
198 ctdb_db->db_name,
199 keep?"storing":"deleting",
200 ctdb_hash(&key)));
201
202 if (keep) {
203 ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
204 } else {
205 ret = tdb_delete(ctdb_db->ltdb->tdb, key);
206 }
207
208 if (ret != 0) {
209 int lvl = DEBUG_ERR;
210
211 if (keep == false &&
212 tdb_error(ctdb_db->ltdb->tdb) == TDB_ERR_NOEXIST)
213 {
214 lvl = DEBUG_DEBUG;
215 }
216
217 DEBUG(lvl, (__location__ " db[%s]: Failed to %s record: "
218 "%d - %s\n",
219 ctdb_db->db_name,
220 keep?"store":"delete", ret,
221 tdb_errorstr(ctdb_db->ltdb->tdb)));
222
223 schedule_for_deletion = false;
224 remove_from_delete_queue = false;
225 }
226 if (seqnum_suppressed) {
227 tdb_add_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
228 }
229
230 talloc_free(rec.dptr);
231
232 if (schedule_for_deletion) {
233 int ret2;
234 ret2 = ctdb_local_schedule_for_deletion(ctdb_db, header, key);
235 if (ret2 != 0) {
236 DEBUG(DEBUG_ERR, (__location__ " ctdb_local_schedule_for_deletion failed.\n"));
237 }
238 }
239
240 if (remove_from_delete_queue) {
241 ctdb_local_remove_from_delete_queue(ctdb_db, header, key);
242 }
243
244 return ret;
245 }
246
247 struct lock_fetch_state {
248 struct ctdb_context *ctdb;
249 void (*recv_pkt)(void *, struct ctdb_req_header *);
250 void *recv_context;
251 struct ctdb_req_header *hdr;
252 uint32_t generation;
253 bool ignore_generation;
254 };
255
256 /*
257 called when we should retry the operation
258 */
259 static void lock_fetch_callback(void *p, bool locked)
260 {
261 struct lock_fetch_state *state = talloc_get_type(p, struct lock_fetch_state);
262 if (!state->ignore_generation &&
263 state->generation != state->ctdb->vnn_map->generation) {
264 DEBUG(DEBUG_NOTICE,("Discarding previous generation lockwait packet\n"));
265 talloc_free(state->hdr);
266 return;
267 }
268 state->recv_pkt(state->recv_context, state->hdr);
269 DEBUG(DEBUG_INFO,(__location__ " PACKET REQUEUED\n"));
270 }
271
272
273 /*
274 do a non-blocking ltdb_lock, deferring this ctdb request until we
275 have the chainlock
276
277 It does the following:
278
279 1) tries to get the chainlock. If it succeeds, then it returns 0
280
281 2) if it fails to get a chainlock immediately then it sets up a
282 non-blocking chainlock via ctdb_lock_record, and when it gets the
283 chainlock it re-submits this ctdb request to the main packet
284 receive function.
285
286 This effectively queues all ctdb requests that cannot be
287 immediately satisfied until it can get the lock. This means that
288 the main ctdb daemon will not block waiting for a chainlock held by
289 a client
290
291 There are 3 possible return values:
292
293 0: means that it got the lock immediately.
294 -1: means that it failed to get the lock, and won't retry
295 -2: means that it failed to get the lock immediately, but will retry
296 */
297 int ctdb_ltdb_lock_requeue(struct ctdb_db_context *ctdb_db,
298 TDB_DATA key, struct ctdb_req_header *hdr,
299 void (*recv_pkt)(void *, struct ctdb_req_header *),
300 void *recv_context, bool ignore_generation)
301 {
302 int ret;
303 struct tdb_context *tdb = ctdb_db->ltdb->tdb;
304 struct lock_request *lreq;
305 struct lock_fetch_state *state;
306
307 ret = tdb_chainlock_nonblock(tdb, key);
308
309 if (ret != 0 &&
310 !(errno == EACCES || errno == EAGAIN || errno == EDEADLK)) {
311 /* a hard failure - don't try again */
312 return -1;
313 }
314
315 /* when torturing, ensure we test the contended path */
316 if ((ctdb_db->ctdb->flags & CTDB_FLAG_TORTURE) &&
317 random() % 5 == 0) {
318 ret = -1;
319 tdb_chainunlock(tdb, key);
320 }
321
322 /* first the non-contended path */
323 if (ret == 0) {
324 return 0;
325 }
326
327 state = talloc(hdr, struct lock_fetch_state);
328 state->ctdb = ctdb_db->ctdb;
329 state->hdr = hdr;
330 state->recv_pkt = recv_pkt;
331 state->recv_context = recv_context;
332 state->generation = ctdb_db->ctdb->vnn_map->generation;
333 state->ignore_generation = ignore_generation;
334
335 /* now the contended path */
336 lreq = ctdb_lock_record(state, ctdb_db, key, true, lock_fetch_callback, state);
337 if (lreq == NULL) {
338 return -1;
339 }
340
341 /* we need to move the packet off the temporary context in ctdb_input_pkt(),
342 so it won't be freed yet */
343 talloc_steal(state, hdr);
344
345 /* now tell the caller than we will retry asynchronously */
346 return -2;
347 }
348
349 /*
350 a varient of ctdb_ltdb_lock_requeue that also fetches the record
351 */
352 int ctdb_ltdb_lock_fetch_requeue(struct ctdb_db_context *ctdb_db,
353 TDB_DATA key, struct ctdb_ltdb_header *header,
354 struct ctdb_req_header *hdr, TDB_DATA *data,
355 void (*recv_pkt)(void *, struct ctdb_req_header *),
356 void *recv_context, bool ignore_generation)
357 {
358 int ret;
359
360 ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr, recv_pkt,
361 recv_context, ignore_generation);
362 if (ret == 0) {
363 ret = ctdb_ltdb_fetch(ctdb_db, key, header, hdr, data);
364 if (ret != 0) {
365 int uret;
366 uret = ctdb_ltdb_unlock(ctdb_db, key);
367 if (uret != 0) {
368 DEBUG(DEBUG_ERR,(__location__ " ctdb_ltdb_unlock() failed with error %d\n", uret));
369 }
370 }
371 }
372 return ret;
373 }
374
375
376 /*
377 paraoid check to see if the db is empty
378 */
379 static void ctdb_check_db_empty(struct ctdb_db_context *ctdb_db)
380 {
381 struct tdb_context *tdb = ctdb_db->ltdb->tdb;
382 int count = tdb_traverse_read(tdb, NULL, NULL);
383 if (count != 0) {
384 DEBUG(DEBUG_ALERT,(__location__ " tdb '%s' not empty on attach! aborting\n",
385 ctdb_db->db_path));
386 ctdb_fatal(ctdb_db->ctdb, "database not empty on attach");
387 }
388 }
389
390 int ctdb_load_persistent_health(struct ctdb_context *ctdb,
391 struct ctdb_db_context *ctdb_db)
392 {
393 struct tdb_context *tdb = ctdb->db_persistent_health->tdb;
394 char *old;
395 char *reason = NULL;
396 TDB_DATA key;
397 TDB_DATA val;
398
399 key.dptr = discard_const_p(uint8_t, ctdb_db->db_name);
400 key.dsize = strlen(ctdb_db->db_name);
401
402 old = ctdb_db->unhealthy_reason;
403 ctdb_db->unhealthy_reason = NULL;
404
405 val = tdb_fetch(tdb, key);
406 if (val.dsize > 0) {
407 reason = talloc_strndup(ctdb_db,
408 (const char *)val.dptr,
409 val.dsize);
410 if (reason == NULL) {
411 DEBUG(DEBUG_ALERT,(__location__ " talloc_strndup(%d) failed\n",
412 (int)val.dsize));
413 ctdb_db->unhealthy_reason = old;
414 free(val.dptr);
415 return -1;
416 }
417 }
418
419 if (val.dptr) {
420 free(val.dptr);
421 }
422
423 talloc_free(old);
424 ctdb_db->unhealthy_reason = reason;
425 return 0;
426 }
427
428 int ctdb_update_persistent_health(struct ctdb_context *ctdb,
429 struct ctdb_db_context *ctdb_db,
430 const char *given_reason,/* NULL means healthy */
431 int num_healthy_nodes)
432 {
433 struct tdb_context *tdb = ctdb->db_persistent_health->tdb;
434 int ret;
435 TDB_DATA key;
436 TDB_DATA val;
437 char *new_reason = NULL;
438 char *old_reason = NULL;
439
440 ret = tdb_transaction_start(tdb);
441 if (ret != 0) {
442 DEBUG(DEBUG_ALERT,(__location__ " tdb_transaction_start('%s') failed: %d - %s\n",
443 tdb_name(tdb), ret, tdb_errorstr(tdb)));
444 return -1;
445 }
446
447 ret = ctdb_load_persistent_health(ctdb, ctdb_db);
448 if (ret != 0) {
449 DEBUG(DEBUG_ALERT,(__location__ " ctdb_load_persistent_health('%s') failed: %d\n",
450 ctdb_db->db_name, ret));
451 return -1;
452 }
453 old_reason = ctdb_db->unhealthy_reason;
454
455 key.dptr = discard_const_p(uint8_t, ctdb_db->db_name);
456 key.dsize = strlen(ctdb_db->db_name);
457
458 if (given_reason) {
459 new_reason = talloc_strdup(ctdb_db, given_reason);
460 if (new_reason == NULL) {
461 DEBUG(DEBUG_ALERT,(__location__ " talloc_strdup(%s) failed\n",
462 given_reason));
463 return -1;
464 }
465 } else if (old_reason && num_healthy_nodes == 0) {
466 /*
467 * If the reason indicates ok, but there where no healthy nodes
468 * available, that it means, we have not recovered valid content
469 * of the db. So if there's an old reason, prefix it with
470 * "NO-HEALTHY-NODES - "
471 */
472 const char *prefix;
473
474 #define _TMP_PREFIX "NO-HEALTHY-NODES - "
475 ret = strncmp(_TMP_PREFIX, old_reason, strlen(_TMP_PREFIX));
476 if (ret != 0) {
477 prefix = _TMP_PREFIX;
478 } else {
479 prefix = "";
480 }
481 new_reason = talloc_asprintf(ctdb_db, "%s%s",
482 prefix, old_reason);
483 if (new_reason == NULL) {
484 DEBUG(DEBUG_ALERT,(__location__ " talloc_asprintf(%s%s) failed\n",
485 prefix, old_reason));
486 return -1;
487 }
488 #undef _TMP_PREFIX
489 }
490
491 if (new_reason) {
492 val.dptr = discard_const_p(uint8_t, new_reason);
493 val.dsize = strlen(new_reason);
494
495 ret = tdb_store(tdb, key, val, TDB_REPLACE);
496 if (ret != 0) {
497 tdb_transaction_cancel(tdb);
498 DEBUG(DEBUG_ALERT,(__location__ " tdb_store('%s', %s, %s) failed: %d - %s\n",
499 tdb_name(tdb), ctdb_db->db_name, new_reason,
500 ret, tdb_errorstr(tdb)));
501 talloc_free(new_reason);
502 return -1;
503 }
504 DEBUG(DEBUG_ALERT,("Updated db health for db(%s) to: %s\n",
505 ctdb_db->db_name, new_reason));
506 } else if (old_reason) {
507 ret = tdb_delete(tdb, key);
508 if (ret != 0) {
509 tdb_transaction_cancel(tdb);
510 DEBUG(DEBUG_ALERT,(__location__ " tdb_delete('%s', %s) failed: %d - %s\n",
511 tdb_name(tdb), ctdb_db->db_name,
512 ret, tdb_errorstr(tdb)));
513 talloc_free(new_reason);
514 return -1;
515 }
516 DEBUG(DEBUG_NOTICE,("Updated db health for db(%s): OK\n",
517 ctdb_db->db_name));
518 }
519
520 ret = tdb_transaction_commit(tdb);
521 if (ret != TDB_SUCCESS) {
522 DEBUG(DEBUG_ALERT,(__location__ " tdb_transaction_commit('%s') failed: %d - %s\n",
523 tdb_name(tdb), ret, tdb_errorstr(tdb)));
524 talloc_free(new_reason);
525 return -1;
526 }
527
528 talloc_free(old_reason);
529 ctdb_db->unhealthy_reason = new_reason;
530
531 return 0;
532 }
533
534 static int ctdb_backup_corrupted_tdb(struct ctdb_context *ctdb,
535 struct ctdb_db_context *ctdb_db)
536 {
537 time_t now = time(NULL);
538 char *new_path;
539 char *new_reason;
540 int ret;
541 struct tm *tm;
542
543 tm = gmtime(&now);
544
545 /* formatted like: foo.tdb.0.corrupted.20091204160825.0Z */
546 new_path = talloc_asprintf(ctdb_db, "%s.corrupted."
547 "%04u%02u%02u%02u%02u%02u.0Z",
548 ctdb_db->db_path,
549 tm->tm_year+1900, tm->tm_mon+1,
550 tm->tm_mday, tm->tm_hour, tm->tm_min,
551 tm->tm_sec);
552 if (new_path == NULL) {
553 DEBUG(DEBUG_CRIT,(__location__ " talloc_asprintf() failed\n"));
554 return -1;
555 }
556
557 new_reason = talloc_asprintf(ctdb_db,
558 "ERROR - Backup of corrupted TDB in '%s'",
559 new_path);
560 if (new_reason == NULL) {
561 DEBUG(DEBUG_CRIT,(__location__ " talloc_asprintf() failed\n"));
562 return -1;
563 }
564 ret = ctdb_update_persistent_health(ctdb, ctdb_db, new_reason, 0);
565 talloc_free(new_reason);
566 if (ret != 0) {
567 DEBUG(DEBUG_CRIT,(__location__
568 ": ctdb_backup_corrupted_tdb(%s) not implemented yet\n",
569 ctdb_db->db_path));
570 return -1;
571 }
572
573 ret = rename(ctdb_db->db_path, new_path);
574 if (ret != 0) {
575 DEBUG(DEBUG_CRIT,(__location__
576 ": ctdb_backup_corrupted_tdb(%s) rename to %s failed: %d - %s\n",
577 ctdb_db->db_path, new_path,
578 errno, strerror(errno)));
579 talloc_free(new_path);
580 return -1;
581 }
582
583 DEBUG(DEBUG_CRIT,(__location__
584 ": ctdb_backup_corrupted_tdb(%s) renamed to %s\n",
585 ctdb_db->db_path, new_path));
586 talloc_free(new_path);
587 return 0;
588 }
589
590 int ctdb_recheck_persistent_health(struct ctdb_context *ctdb)
591 {
592 struct ctdb_db_context *ctdb_db;
593 int ret;
594 int ok = 0;
595 int fail = 0;
596
597 for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
598 if (!ctdb_db->persistent) {
599 continue;
600 }
601
602 ret = ctdb_load_persistent_health(ctdb, ctdb_db);
603 if (ret != 0) {
604 DEBUG(DEBUG_ALERT,(__location__
605 " load persistent health for '%s' failed\n",
606 ctdb_db->db_path));
607 return -1;
608 }
609
610 if (ctdb_db->unhealthy_reason == NULL) {
611 ok++;
612 DEBUG(DEBUG_INFO,(__location__
613 " persistent db '%s' healthy\n",
614 ctdb_db->db_path));
615 continue;
616 }
617
618 fail++;
619 DEBUG(DEBUG_ALERT,(__location__
620 " persistent db '%s' unhealthy: %s\n",
621 ctdb_db->db_path,
622 ctdb_db->unhealthy_reason));
623 }
624 DEBUG((fail!=0)?DEBUG_ALERT:DEBUG_NOTICE,
625 ("ctdb_recheck_presistent_health: OK[%d] FAIL[%d]\n",
626 ok, fail));
627
628 if (fail != 0) {
629 return -1;
630 }
631
632 return 0;
633 }
634
635
636 /*
637 mark a database - as healthy
638 */
639 int32_t ctdb_control_db_set_healthy(struct ctdb_context *ctdb, TDB_DATA indata)
640 {
641 uint32_t db_id = *(uint32_t *)indata.dptr;
642 struct ctdb_db_context *ctdb_db;
643 int ret;
644 bool may_recover = false;
645
646 ctdb_db = find_ctdb_db(ctdb, db_id);
647 if (!ctdb_db) {
648 DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%x\n", db_id));
649 return -1;
650 }
651
652 if (ctdb_db->unhealthy_reason) {
653 may_recover = true;
654 }
655
656 ret = ctdb_update_persistent_health(ctdb, ctdb_db, NULL, 1);
657 if (ret != 0) {
658 DEBUG(DEBUG_ERR,(__location__
659 " ctdb_update_persistent_health(%s) failed\n",
660 ctdb_db->db_name));
661 return -1;
662 }
663
664 if (may_recover && ctdb->runstate == CTDB_RUNSTATE_STARTUP) {
665 DEBUG(DEBUG_ERR, (__location__ " db %s become healthy - force recovery for startup\n",
666 ctdb_db->db_name));
667 ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
668 }
669
670 return 0;
671 }
672
673 int32_t ctdb_control_db_get_health(struct ctdb_context *ctdb,
674 TDB_DATA indata,
675 TDB_DATA *outdata)
676 {
677 uint32_t db_id = *(uint32_t *)indata.dptr;
678 struct ctdb_db_context *ctdb_db;
679 int ret;
680
681 ctdb_db = find_ctdb_db(ctdb, db_id);
682 if (!ctdb_db) {
683 DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%x\n", db_id));
684 return -1;
685 }
686
687 ret = ctdb_load_persistent_health(ctdb, ctdb_db);
688 if (ret != 0) {
689 DEBUG(DEBUG_ERR,(__location__
690 " ctdb_load_persistent_health(%s) failed\n",
691 ctdb_db->db_name));
692 return -1;
693 }
694
695 *outdata = tdb_null;
696 if (ctdb_db->unhealthy_reason) {
697 outdata->dptr = (uint8_t *)ctdb_db->unhealthy_reason;
698 outdata->dsize = strlen(ctdb_db->unhealthy_reason)+1;
699 }
700
701 return 0;
702 }
703
704
705 int ctdb_set_db_readonly(struct ctdb_context *ctdb, struct ctdb_db_context *ctdb_db)
706 {
707 char *ropath;
708
709 if (ctdb_db->readonly) {
710 return 0;
711 }
712
713 if (ctdb_db->persistent) {
714 DEBUG(DEBUG_ERR,("Persistent databases do not support readonly property\n"));
715 return -1;
716 }
717
718 ropath = talloc_asprintf(ctdb_db, "%s.RO", ctdb_db->db_path);
719 if (ropath == NULL) {
720 DEBUG(DEBUG_CRIT,("Failed to asprintf the tracking database\n"));
721 return -1;
722 }
723 ctdb_db->rottdb = tdb_open(ropath,
724 ctdb->tunable.database_hash_size,
725 TDB_NOLOCK|TDB_CLEAR_IF_FIRST|TDB_NOSYNC,
726 O_CREAT|O_RDWR, 0);
727 if (ctdb_db->rottdb == NULL) {
728 DEBUG(DEBUG_CRIT,("Failed to open/create the tracking database '%s'\n", ropath));
729 talloc_free(ropath);
730 return -1;
731 }
732
733 DEBUG(DEBUG_NOTICE,("OPENED tracking database : '%s'\n", ropath));
734
735 ctdb_db->readonly = true;
736
737 DEBUG(DEBUG_NOTICE, ("Readonly property set on DB %s\n", ctdb_db->db_name));
738
739 talloc_free(ropath);
740 return 0;
741 }
742
743 /*
744 attach to a database, handling both persistent and non-persistent databases
745 return 0 on success, -1 on failure
746 */
747 static int ctdb_local_attach(struct ctdb_context *ctdb, const char *db_name,
748 bool persistent, const char *unhealthy_reason,
749 bool jenkinshash, bool mutexes)
750 {
751 struct ctdb_db_context *ctdb_db, *tmp_db;
752 int ret;
753 struct TDB_DATA key;
754 unsigned tdb_flags;
755 int mode = 0600;
756 int remaining_tries = 0;
757
758 ctdb_db = talloc_zero(ctdb, struct ctdb_db_context);
759 CTDB_NO_MEMORY(ctdb, ctdb_db);
760
761 ctdb_db->priority = 1;
762 ctdb_db->ctdb = ctdb;
763 ctdb_db->db_name = talloc_strdup(ctdb_db, db_name);
764 CTDB_NO_MEMORY(ctdb, ctdb_db->db_name);
765
766 key.dsize = strlen(db_name)+1;
767 key.dptr = discard_const(db_name);
768 ctdb_db->db_id = ctdb_hash(&key);
769 ctdb_db->persistent = persistent;
770
771 if (!ctdb_db->persistent) {
772 ctdb_db->delete_queue = trbt_create(ctdb_db, 0);
773 if (ctdb_db->delete_queue == NULL) {
774 CTDB_NO_MEMORY(ctdb, ctdb_db->delete_queue);
775 }
776
777 ctdb_db->ctdb_ltdb_store_fn = ctdb_ltdb_store_server;
778 }
779
780 /* check for hash collisions */
781 for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
782 if (tmp_db->db_id == ctdb_db->db_id) {
783 DEBUG(DEBUG_CRIT,("db_id 0x%x hash collision. name1='%s' name2='%s'\n",
784 tmp_db->db_id, db_name, tmp_db->db_name));
785 talloc_free(ctdb_db);
786 return -1;
787 }
788 }
789
790 if (persistent) {
791 if (unhealthy_reason) {
792 ret = ctdb_update_persistent_health(ctdb, ctdb_db,
793 unhealthy_reason, 0);
794 if (ret != 0) {
795 DEBUG(DEBUG_ALERT,(__location__ " ctdb_update_persistent_health('%s','%s') failed: %d\n",
796 ctdb_db->db_name, unhealthy_reason, ret));
797 talloc_free(ctdb_db);
798 return -1;
799 }
800 }
801
802 if (ctdb->max_persistent_check_errors > 0) {
803 remaining_tries = 1;
804 }
805 if (ctdb->runstate == CTDB_RUNSTATE_RUNNING) {
806 remaining_tries = 0;
807 }
808
809 ret = ctdb_load_persistent_health(ctdb, ctdb_db);
810 if (ret != 0) {
811 DEBUG(DEBUG_ALERT,(__location__ " ctdb_load_persistent_health('%s') failed: %d\n",
812 ctdb_db->db_name, ret));
813 talloc_free(ctdb_db);
814 return -1;
815 }
816 }
817
818 if (ctdb_db->unhealthy_reason && remaining_tries == 0) {
819 DEBUG(DEBUG_ALERT,(__location__ "ERROR: tdb %s is marked as unhealthy: %s\n",
820 ctdb_db->db_name, ctdb_db->unhealthy_reason));
821 talloc_free(ctdb_db);
822 return -1;
823 }
824
825 if (ctdb_db->unhealthy_reason) {
826 /* this is just a warning, but we want that in the log file! */
827 DEBUG(DEBUG_ALERT,(__location__ "Warning: tdb %s is marked as unhealthy: %s\n",
828 ctdb_db->db_name, ctdb_db->unhealthy_reason));
829 }
830
831 /* open the database */
832 ctdb_db->db_path = talloc_asprintf(ctdb_db, "%s/%s.%u",
833 persistent?ctdb->db_directory_persistent:ctdb->db_directory,
834 db_name, ctdb->pnn);
835
836 tdb_flags = persistent? TDB_DEFAULT : TDB_CLEAR_IF_FIRST | TDB_NOSYNC;
837 if (ctdb->valgrinding) {
838 tdb_flags |= TDB_NOMMAP;
839 }
840 tdb_flags |= TDB_DISALLOW_NESTING;
841 if (jenkinshash) {
842 tdb_flags |= TDB_INCOMPATIBLE_HASH;
843 }
844 #ifdef TDB_MUTEX_LOCKING
845 if (ctdb->tunable.mutex_enabled && mutexes &&
846 tdb_runtime_check_for_robust_mutexes()) {
847 tdb_flags |= (TDB_MUTEX_LOCKING | TDB_CLEAR_IF_FIRST);
848 }
849 #endif
850
851 again:
852 ctdb_db->ltdb = tdb_wrap_open(ctdb_db, ctdb_db->db_path,
853 ctdb->tunable.database_hash_size,
854 tdb_flags,
855 O_CREAT|O_RDWR, mode);
856 if (ctdb_db->ltdb == NULL) {
857 struct stat st;
858 int saved_errno = errno;
859
860 if (!persistent) {
861 DEBUG(DEBUG_CRIT,("Failed to open tdb '%s': %d - %s\n",
862 ctdb_db->db_path,
863 saved_errno,
864 strerror(saved_errno)));
865 talloc_free(ctdb_db);
866 return -1;
867 }
868
869 if (remaining_tries == 0) {
870 DEBUG(DEBUG_CRIT,(__location__
871 "Failed to open persistent tdb '%s': %d - %s\n",
872 ctdb_db->db_path,
873 saved_errno,
874 strerror(saved_errno)));
875 talloc_free(ctdb_db);
876 return -1;
877 }
878
879 ret = stat(ctdb_db->db_path, &st);
880 if (ret != 0) {
881 DEBUG(DEBUG_CRIT,(__location__
882 "Failed to open persistent tdb '%s': %d - %s\n",
883 ctdb_db->db_path,
884 saved_errno,
885 strerror(saved_errno)));
886 talloc_free(ctdb_db);
887 return -1;
888 }
889
890 ret = ctdb_backup_corrupted_tdb(ctdb, ctdb_db);
891 if (ret != 0) {
892 DEBUG(DEBUG_CRIT,(__location__
893 "Failed to open persistent tdb '%s': %d - %s\n",
894 ctdb_db->db_path,
895 saved_errno,
896 strerror(saved_errno)));
897 talloc_free(ctdb_db);
898 return -1;
899 }
900
901 remaining_tries--;
902 mode = st.st_mode;
903 goto again;
904 }
905
906 if (!persistent) {
907 ctdb_check_db_empty(ctdb_db);
908 } else {
909 ret = tdb_check(ctdb_db->ltdb->tdb, NULL, NULL);
910 if (ret != 0) {
911 int fd;
912 struct stat st;
913
914 DEBUG(DEBUG_CRIT,("tdb_check(%s) failed: %d - %s\n",
915 ctdb_db->db_path, ret,
916 tdb_errorstr(ctdb_db->ltdb->tdb)));
917 if (remaining_tries == 0) {
918 talloc_free(ctdb_db);
919 return -1;
920 }
921
922 fd = tdb_fd(ctdb_db->ltdb->tdb);
923 ret = fstat(fd, &st);
924 if (ret != 0) {
925 DEBUG(DEBUG_CRIT,(__location__
926 "Failed to fstat() persistent tdb '%s': %d - %s\n",
927 ctdb_db->db_path,
928 errno,
929 strerror(errno)));
930 talloc_free(ctdb_db);
931 return -1;
932 }
933
934 /* close the TDB */
935 talloc_free(ctdb_db->ltdb);
936 ctdb_db->ltdb = NULL;
937
938 ret = ctdb_backup_corrupted_tdb(ctdb, ctdb_db);
939 if (ret != 0) {
940 DEBUG(DEBUG_CRIT,("Failed to backup corrupted tdb '%s'\n",
941 ctdb_db->db_path));
942 talloc_free(ctdb_db);
943 return -1;
944 }
945
946 remaining_tries--;
947 mode = st.st_mode;
948 goto again;
949 }
950 }
951
952 /* set up a rb tree we can use to track which records we have a
953 fetch-lock in-flight for so we can defer any additional calls
954 for the same record.
955 */
956 ctdb_db->deferred_fetch = trbt_create(ctdb_db, 0);
957 if (ctdb_db->deferred_fetch == NULL) {
958 DEBUG(DEBUG_ERR,("Failed to create deferred fetch rb tree for ctdb database\n"));
959 talloc_free(ctdb_db);
960 return -1;
961 }
962
963 ctdb_db->defer_dmaster = trbt_create(ctdb_db, 0);
964 if (ctdb_db->defer_dmaster == NULL) {
965 DEBUG(DEBUG_ERR, ("Failed to create defer dmaster rb tree for %s\n",
966 ctdb_db->db_name));
967 talloc_free(ctdb_db);
968 return -1;
969 }
970
971 DLIST_ADD(ctdb->db_list, ctdb_db);
972
973 /* setting this can help some high churn databases */
974 tdb_set_max_dead(ctdb_db->ltdb->tdb, ctdb->tunable.database_max_dead);
975
976 /*
977 all databases support the "null" function. we need this in
978 order to do forced migration of records
979 */
980 ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_null_func, CTDB_NULL_FUNC);
981 if (ret != 0) {
982 DEBUG(DEBUG_CRIT,("Failed to setup null function for '%s'\n", ctdb_db->db_name));
983 talloc_free(ctdb_db);
984 return -1;
985 }
986
987 /*
988 all databases support the "fetch" function. we need this
989 for efficient Samba3 ctdb fetch
990 */
991 ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_fetch_func, CTDB_FETCH_FUNC);
992 if (ret != 0) {
993 DEBUG(DEBUG_CRIT,("Failed to setup fetch function for '%s'\n", ctdb_db->db_name));
994 talloc_free(ctdb_db);
995 return -1;
996 }
997
998 /*
999 all databases support the "fetch_with_header" function. we need this
1000 for efficient readonly record fetches
1001 */
1002 ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_fetch_with_header_func, CTDB_FETCH_WITH_HEADER_FUNC);
1003 if (ret != 0) {
1004 DEBUG(DEBUG_CRIT,("Failed to setup fetch function for '%s'\n", ctdb_db->db_name));
1005 talloc_free(ctdb_db);
1006 return -1;
1007 }
1008
1009 ret = ctdb_vacuum_init(ctdb_db);
1010 if (ret != 0) {
1011 DEBUG(DEBUG_CRIT,("Failed to setup vacuuming for "
1012 "database '%s'\n", ctdb_db->db_name));
1013 talloc_free(ctdb_db);
1014 return -1;
1015 }
1016
1017
1018 DEBUG(DEBUG_NOTICE,("Attached to database '%s' with flags 0x%x\n",
1019 ctdb_db->db_path, tdb_flags));
1020
1021 /* success */
1022 return 0;
1023 }
1024
1025
1026 struct ctdb_deferred_attach_context {
1027 struct ctdb_deferred_attach_context *next, *prev;
1028 struct ctdb_context *ctdb;
1029 struct ctdb_req_control *c;
1030 };
1031
1032
1033 static int ctdb_deferred_attach_destructor(struct ctdb_deferred_attach_context *da_ctx)
1034 {
1035 DLIST_REMOVE(da_ctx->ctdb->deferred_attach, da_ctx);
1036
1037 return 0;
1038 }
1039
1040 static void ctdb_deferred_attach_timeout(struct event_context *ev, struct timed_event *te, struct timeval t, void *private_data)
1041 {
1042 struct ctdb_deferred_attach_context *da_ctx = talloc_get_type(private_data, struct ctdb_deferred_attach_context);
1043 struct ctdb_context *ctdb = da_ctx->ctdb;
1044
1045 ctdb_request_control_reply(ctdb, da_ctx->c, NULL, -1, NULL);
1046 talloc_free(da_ctx);
1047 }
1048
1049 static void ctdb_deferred_attach_callback(struct event_context *ev, struct timed_event *te, struct timeval t, void *private_data)
1050 {
1051 struct ctdb_deferred_attach_context *da_ctx = talloc_get_type(private_data, struct ctdb_deferred_attach_context);
1052 struct ctdb_context *ctdb = da_ctx->ctdb;
1053
1054 /* This talloc-steals the packet ->c */
1055 ctdb_input_pkt(ctdb, (struct ctdb_req_header *)da_ctx->c);
1056 talloc_free(da_ctx);
1057 }
1058
1059 int ctdb_process_deferred_attach(struct ctdb_context *ctdb)
1060 {
1061 struct ctdb_deferred_attach_context *da_ctx;
1062
1063 /* call it from the main event loop as soon as the current event
1064 finishes.
1065 */
1066 while ((da_ctx = ctdb->deferred_attach) != NULL) {
1067 DLIST_REMOVE(ctdb->deferred_attach, da_ctx);
1068 event_add_timed(ctdb->ev, da_ctx, timeval_current_ofs(1,0), ctdb_deferred_attach_callback, da_ctx);
1069 }
1070
1071 return 0;
1072 }
1073
1074 /*
1075 a client has asked to attach a new database
1076 */
1077 int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
1078 TDB_DATA *outdata, uint64_t tdb_flags,
1079 bool persistent, uint32_t client_id,
1080 struct ctdb_req_control *c,
1081 bool *async_reply)
1082 {
1083 const char *db_name = (const char *)indata.dptr;
1084 struct ctdb_db_context *db;
1085 struct ctdb_node *node = ctdb->nodes[ctdb->pnn];
1086 struct ctdb_client *client = NULL;
1087 bool with_jenkinshash, with_mutexes;
1088
1089 if (ctdb->tunable.allow_client_db_attach == 0) {
1090 DEBUG(DEBUG_ERR, ("DB Attach to database %s denied by tunable "
1091 "AllowClientDBAccess == 0\n", db_name));
1092 return -1;
1093 }
1094
1095 /* dont allow any local clients to attach while we are in recovery mode
1096 * except for the recovery daemon.
1097 * allow all attach from the network since these are always from remote
1098 * recovery daemons.
1099 */
1100 if (client_id != 0) {
1101 client = ctdb_reqid_find(ctdb, client_id, struct ctdb_client);
1102 }
1103 if (client != NULL) {
1104 /* If the node is inactive it is not part of the cluster
1105 and we should not allow clients to attach to any
1106 databases
1107 */
1108 if (node->flags & NODE_FLAGS_INACTIVE) {
1109 DEBUG(DEBUG_ERR,("DB Attach to database %s refused since node is inactive (flags=0x%x)\n", db_name, node->flags));
1110 return -1;
1111 }
1112
1113 if (ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE &&
1114 client->pid != ctdb->recoverd_pid &&
1115 ctdb->runstate < CTDB_RUNSTATE_RUNNING) {
1116 struct ctdb_deferred_attach_context *da_ctx = talloc(client, struct ctdb_deferred_attach_context);
1117
1118 if (da_ctx == NULL) {
1119 DEBUG(DEBUG_ERR,("DB Attach to database %s deferral for client with pid:%d failed due to OOM.\n", db_name, client->pid));
1120 return -1;
1121 }
1122
1123 da_ctx->ctdb = ctdb;
1124 da_ctx->c = talloc_steal(da_ctx, c);
1125 talloc_set_destructor(da_ctx, ctdb_deferred_attach_destructor);
1126 DLIST_ADD(ctdb->deferred_attach, da_ctx);
1127
1128 event_add_timed(ctdb->ev, da_ctx, timeval_current_ofs(ctdb->tunable.deferred_attach_timeout, 0), ctdb_deferred_attach_timeout, da_ctx);
1129
1130 DEBUG(DEBUG_ERR,("DB Attach to database %s deferred for client with pid:%d since node is in recovery mode.\n", db_name, client->pid));
1131 *async_reply = true;
1132 return 0;
1133 }
1134 }
1135
1136 /* the client can optionally pass additional tdb flags, but we
1137 only allow a subset of those on the database in ctdb. Note
1138 that tdb_flags is passed in via the (otherwise unused)
1139 srvid to the attach control */
1140 #ifdef TDB_MUTEX_LOCKING
1141 tdb_flags &= (TDB_NOSYNC|TDB_INCOMPATIBLE_HASH|TDB_MUTEX_LOCKING|TDB_CLEAR_IF_FIRST);
1142 #else
1143 tdb_flags &= (TDB_NOSYNC|TDB_INCOMPATIBLE_HASH);
1144 #endif
1145
1146 /* see if we already have this name */
1147 db = ctdb_db_handle(ctdb, db_name);
1148 if (db) {
1149 if (db->persistent != persistent) {
1150 DEBUG(DEBUG_ERR, ("ERROR: DB Attach %spersistent to %spersistent "
1151 "database %s\n", persistent ? "" : "non-",
1152 db-> persistent ? "" : "non-", db_name));
1153 return -1;
1154 }
1155 outdata->dptr = (uint8_t *)&db->db_id;
1156 outdata->dsize = sizeof(db->db_id);
1157 tdb_add_flags(db->ltdb->tdb, tdb_flags);
1158 return 0;
1159 }
1160
1161 with_jenkinshash = (tdb_flags & TDB_INCOMPATIBLE_HASH) ? true : false;
1162 #ifdef TDB_MUTEX_LOCKING
1163 with_mutexes = (tdb_flags & TDB_MUTEX_LOCKING) ? true : false;
1164 #else
1165 with_mutexes = false;
1166 #endif
1167
1168 if (ctdb_local_attach(ctdb, db_name, persistent, NULL,
1169 with_jenkinshash, with_mutexes) != 0) {
1170 return -1;
1171 }
1172
1173 db = ctdb_db_handle(ctdb, db_name);
1174 if (!db) {
1175 DEBUG(DEBUG_ERR,("Failed to find db handle for name '%s'\n", db_name));
1176 return -1;
1177 }
1178
1179 /* remember the flags the client has specified */
1180 tdb_add_flags(db->ltdb->tdb, tdb_flags);
1181
1182 outdata->dptr = (uint8_t *)&db->db_id;
1183 outdata->dsize = sizeof(db->db_id);
1184
1185 /* Try to ensure it's locked in mem */
1186 lockdown_memory(ctdb->valgrinding);
1187
1188 /* tell all the other nodes about this database */
1189 ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_ALL, tdb_flags,
1190 persistent?CTDB_CONTROL_DB_ATTACH_PERSISTENT:
1191 CTDB_CONTROL_DB_ATTACH,
1192 0, CTDB_CTRL_FLAG_NOREPLY,
1193 indata, NULL, NULL);
1194
1195 /* success */
1196 return 0;
1197 }
1198
1199 /*
1200 * a client has asked to detach from a database
1201 */
1202 int32_t ctdb_control_db_detach(struct ctdb_context *ctdb, TDB_DATA indata,
1203 uint32_t client_id)
1204 {
1205 uint32_t db_id;
1206 struct ctdb_db_context *ctdb_db;
1207 struct ctdb_client *client = NULL;
1208
1209 db_id = *(uint32_t *)indata.dptr;
1210 ctdb_db = find_ctdb_db(ctdb, db_id);
1211 if (ctdb_db == NULL) {
1212 DEBUG(DEBUG_ERR, ("Invalid dbid 0x%08x in DB detach\n",
1213 db_id));
1214 return -1;
1215 }
1216
1217 if (ctdb->tunable.allow_client_db_attach == 1) {
1218 DEBUG(DEBUG_ERR, ("DB detach from database %s denied. "
1219 "Clients are allowed access to databases "
1220 "(AllowClientDBAccess == 1)\n",
1221 ctdb_db->db_name));
1222 return -1;
1223 }
1224
1225 if (ctdb_db->persistent) {
1226 DEBUG(DEBUG_ERR, ("DB detach from persistent database %s "
1227 "denied\n", ctdb_db->db_name));
1228 return -1;
1229 }
1230
1231 /* Cannot detach from database when in recovery */
1232 if (ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE) {
1233 DEBUG(DEBUG_ERR, ("DB detach denied while in recovery\n"));
1234 return -1;
1235 }
1236
1237 /* If a control comes from a client, then broadcast it to all nodes.
1238 * Do the actual detach only if the control comes from other daemons.
1239 */
1240 if (client_id != 0) {
1241 client = ctdb_reqid_find(ctdb, client_id, struct ctdb_client);
1242 if (client != NULL) {
1243 /* forward the control to all the nodes */
1244 ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_ALL, 0,
1245 CTDB_CONTROL_DB_DETACH, 0,
1246 CTDB_CTRL_FLAG_NOREPLY,
1247 indata, NULL, NULL);
1248 return 0;
1249 }
1250 DEBUG(DEBUG_ERR, ("Client has gone away. Failing DB detach "
1251 "for database '%s'\n", ctdb_db->db_name));
1252 return -1;
1253 }
1254
1255 /* Detach database from recoverd */
1256 if (ctdb_daemon_send_message(ctdb, ctdb->pnn,
1257 CTDB_SRVID_DETACH_DATABASE,
1258 indata) != 0) {
1259 DEBUG(DEBUG_ERR, ("Unable to detach DB from recoverd\n"));
1260 return -1;
1261 }
1262
1263 /* Disable vacuuming and drop all vacuuming data */
1264 talloc_free(ctdb_db->vacuum_handle);
1265 talloc_free(ctdb_db->delete_queue);
1266
1267 /* Terminate any deferred fetch */
1268 talloc_free(ctdb_db->deferred_fetch);
1269
1270 /* Terminate any traverses */
1271 while (ctdb_db->traverse) {
1272 talloc_free(ctdb_db->traverse);
1273 }
1274
1275 /* Terminate any revokes */
1276 while (ctdb_db->revokechild_active) {
1277 talloc_free(ctdb_db->revokechild_active);
1278 }
1279
1280 /* Free readonly tracking database */
1281 if (ctdb_db->readonly) {
1282 talloc_free(ctdb_db->rottdb);
1283 }
1284
1285 DLIST_REMOVE(ctdb->db_list, ctdb_db);
1286
1287 DEBUG(DEBUG_NOTICE, ("Detached from database '%s'\n",
1288 ctdb_db->db_name));
1289 talloc_free(ctdb_db);
1290
1291 return 0;
1292 }
1293
1294 /*
1295 attach to all existing persistent databases
1296 */
1297 static int ctdb_attach_persistent(struct ctdb_context *ctdb,
1298 const char *unhealthy_reason)
1299 {
1300 DIR *d;
1301 struct dirent *de;
1302
1303 /* open the persistent db directory and scan it for files */
1304 d = opendir(ctdb->db_directory_persistent);
1305 if (d == NULL) {
1306 return 0;
1307 }
1308
1309 while ((de=readdir(d))) {
1310 char *p, *s, *q;
1311 size_t len = strlen(de->d_name);
1312 uint32_t node;
1313 int invalid_name = 0;
1314
1315 s = talloc_strdup(ctdb, de->d_name);
1316 if (s == NULL) {
1317 closedir(d);
1318 CTDB_NO_MEMORY(ctdb, s);
1319 }
1320
1321 /* only accept names ending in .tdb */
1322 p = strstr(s, ".tdb.");
1323 if (len < 7 || p == NULL) {
1324 talloc_free(s);
1325 continue;
1326 }
1327
1328 /* only accept names ending with .tdb. and any number of digits */
1329 q = p+5;
1330 while (*q != 0 && invalid_name == 0) {
1331 if (!isdigit(*q++)) {
1332 invalid_name = 1;
1333 }
1334 }
1335 if (invalid_name == 1 || sscanf(p+5, "%u", &node) != 1 || node != ctdb->pnn) {
1336 DEBUG(DEBUG_ERR,("Ignoring persistent database '%s'\n", de->d_name));
1337 talloc_free(s);
1338 continue;
1339 }
1340 p[4] = 0;
1341
1342 if (ctdb_local_attach(ctdb, s, true, unhealthy_reason, false, false) != 0) {
1343 DEBUG(DEBUG_ERR,("Failed to attach to persistent database '%s'\n", de->d_name));
1344 closedir(d);
1345 talloc_free(s);
1346 return -1;
1347 }
1348
1349 DEBUG(DEBUG_INFO,("Attached to persistent database %s\n", s));
1350
1351 talloc_free(s);
1352 }
1353 closedir(d);
1354 return 0;
1355 }
1356
1357 int ctdb_attach_databases(struct ctdb_context *ctdb)
1358 {
1359 int ret;
1360 char *persistent_health_path = NULL;
1361 char *unhealthy_reason = NULL;
1362 bool first_try = true;
1363
1364 persistent_health_path = talloc_asprintf(ctdb, "%s/%s.%u",
1365 ctdb->db_directory_state,
1366 PERSISTENT_HEALTH_TDB,
1367 ctdb->pnn);
1368 if (persistent_health_path == NULL) {
1369 DEBUG(DEBUG_CRIT,(__location__ " talloc_asprintf() failed\n"));
1370 return -1;
1371 }
1372
1373 again:
1374
1375 ctdb->db_persistent_health = tdb_wrap_open(ctdb, persistent_health_path,
1376 0, TDB_DISALLOW_NESTING,
1377 O_CREAT | O_RDWR, 0600);
1378 if (ctdb->db_persistent_health == NULL) {
1379 struct tdb_wrap *tdb;
1380
1381 if (!first_try) {
1382 DEBUG(DEBUG_CRIT,("Failed to open tdb '%s': %d - %s\n",
1383 persistent_health_path,
1384 errno,
1385 strerror(errno)));
1386 talloc_free(persistent_health_path);
1387 talloc_free(unhealthy_reason);
1388 return -1;
1389 }
1390 first_try = false;
1391
1392 unhealthy_reason = talloc_asprintf(ctdb, "WARNING - '%s' %s - %s",
1393 persistent_health_path,
1394 "was cleared after a failure",
1395 "manual verification needed");
1396 if (unhealthy_reason == NULL) {
1397 DEBUG(DEBUG_CRIT,(__location__ " talloc_asprintf() failed\n"));
1398 talloc_free(persistent_health_path);
1399 return -1;
1400 }
1401
1402 DEBUG(DEBUG_CRIT,("Failed to open tdb '%s' - retrying after CLEAR_IF_FIRST\n",
1403 persistent_health_path));
1404 tdb = tdb_wrap_open(ctdb, persistent_health_path,
1405 0, TDB_CLEAR_IF_FIRST | TDB_DISALLOW_NESTING,
1406 O_CREAT | O_RDWR, 0600);
1407 if (tdb) {
1408 DEBUG(DEBUG_CRIT,("Failed to open tdb '%s' - with CLEAR_IF_FIRST: %d - %s\n",
1409 persistent_health_path,
1410 errno,
1411 strerror(errno)));
1412 talloc_free(persistent_health_path);
1413 talloc_free(unhealthy_reason);
1414 return -1;
1415 }
1416
1417 talloc_free(tdb);
1418 goto again;
1419 }
1420 ret = tdb_check(ctdb->db_persistent_health->tdb, NULL, NULL);
1421 if (ret != 0) {
1422 struct tdb_wrap *tdb;
1423
1424 talloc_free(ctdb->db_persistent_health);
1425 ctdb->db_persistent_health = NULL;
1426
1427 if (!first_try) {
1428 DEBUG(DEBUG_CRIT,("tdb_check('%s') failed\n",
1429 persistent_health_path));
1430 talloc_free(persistent_health_path);
1431 talloc_free(unhealthy_reason);
1432 return -1;
1433 }
1434 first_try = false;
1435
1436 unhealthy_reason = talloc_asprintf(ctdb, "WARNING - '%s' %s - %s",
1437 persistent_health_path,
1438 "was cleared after a failure",
1439 "manual verification needed");
1440 if (unhealthy_reason == NULL) {
1441 DEBUG(DEBUG_CRIT,(__location__ " talloc_asprintf() failed\n"));
1442 talloc_free(persistent_health_path);
1443 return -1;
1444 }
1445
1446 DEBUG(DEBUG_CRIT,("tdb_check('%s') failed - retrying after CLEAR_IF_FIRST\n",
1447 persistent_health_path));
1448 tdb = tdb_wrap_open(ctdb, persistent_health_path,
1449 0, TDB_CLEAR_IF_FIRST | TDB_DISALLOW_NESTING,
1450 O_CREAT | O_RDWR, 0600);
1451 if (tdb) {
1452 DEBUG(DEBUG_CRIT,("Failed to open tdb '%s' - with CLEAR_IF_FIRST: %d - %s\n",
1453 persistent_health_path,
1454 errno,
1455 strerror(errno)));
1456 talloc_free(persistent_health_path);
1457 talloc_free(unhealthy_reason);
1458 return -1;
1459 }
1460
1461 talloc_free(tdb);
1462 goto again;
1463 }
1464 talloc_free(persistent_health_path);
1465
1466 ret = ctdb_attach_persistent(ctdb, unhealthy_reason);
1467 talloc_free(unhealthy_reason);
1468 if (ret != 0) {
1469 return ret;
1470 }
1471
1472 return 0;
1473 }
1474
1475 /*
1476 called when a broadcast seqnum update comes in
1477 */
1478 int32_t ctdb_ltdb_update_seqnum(struct ctdb_context *ctdb, uint32_t db_id, uint32_t srcnode)
1479 {
1480 struct ctdb_db_context *ctdb_db;
1481 if (srcnode == ctdb->pnn) {
1482 /* don't update ourselves! */
1483 return 0;
1484 }
1485
1486 ctdb_db = find_ctdb_db(ctdb, db_id);
1487 if (!ctdb_db) {
1488 DEBUG(DEBUG_ERR,("Unknown db_id 0x%x in ctdb_ltdb_update_seqnum\n", db_id));
1489 return -1;
1490 }
1491
1492 if (ctdb_db->unhealthy_reason) {
1493 DEBUG(DEBUG_ERR,("db(%s) unhealty in ctdb_ltdb_update_seqnum: %s\n",
1494 ctdb_db->db_name, ctdb_db->unhealthy_reason));
1495 return -1;
1496 }
1497
1498 tdb_increment_seqnum_nonblock(ctdb_db->ltdb->tdb);
1499 ctdb_db->seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
1500 return 0;
1501 }
1502
1503 /*
1504 timer to check for seqnum changes in a ltdb and propogate them
1505 */
1506 static void ctdb_ltdb_seqnum_check(struct event_context *ev, struct timed_event *te,
1507 struct timeval t, void *p)
1508 {
1509 struct ctdb_db_context *ctdb_db = talloc_get_type(p, struct ctdb_db_context);
1510 struct ctdb_context *ctdb = ctdb_db->ctdb;
1511 uint32_t new_seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
1512 if (new_seqnum != ctdb_db->seqnum) {
1513 /* something has changed - propogate it */
1514 TDB_DATA data;
1515 data.dptr = (uint8_t *)&ctdb_db->db_id;
1516 data.dsize = sizeof(uint32_t);
1517 ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_VNNMAP, 0,
1518 CTDB_CONTROL_UPDATE_SEQNUM, 0, CTDB_CTRL_FLAG_NOREPLY,
1519 data, NULL, NULL);
1520 }
1521 ctdb_db->seqnum = new_seqnum;
1522
1523 /* setup a new timer */
1524 ctdb_db->seqnum_update =
1525 event_add_timed(ctdb->ev, ctdb_db,
1526 timeval_current_ofs(ctdb->tunable.seqnum_interval/1000, (ctdb->tunable.seqnum_interval%1000)*1000),
1527 ctdb_ltdb_seqnum_check, ctdb_db);
1528 }
1529
1530 /*
1531 enable seqnum handling on this db
1532 */
1533 int32_t ctdb_ltdb_enable_seqnum(struct ctdb_context *ctdb, uint32_t db_id)
1534 {
1535 struct ctdb_db_context *ctdb_db;
1536 ctdb_db = find_ctdb_db(ctdb, db_id);
1537 if (!ctdb_db) {
1538 DEBUG(DEBUG_ERR,("Unknown db_id 0x%x in ctdb_ltdb_enable_seqnum\n", db_id));
1539 return -1;
1540 }
1541
1542 if (ctdb_db->seqnum_update == NULL) {
1543 ctdb_db->seqnum_update =
1544 event_add_timed(ctdb->ev, ctdb_db,
1545 timeval_current_ofs(ctdb->tunable.seqnum_interval/1000, (ctdb->tunable.seqnum_interval%1000)*1000),
1546 ctdb_ltdb_seqnum_check, ctdb_db);
1547 }
1548
1549 tdb_enable_seqnum(ctdb_db->ltdb->tdb);
1550 ctdb_db->seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
1551 return 0;
1552 }
1553
1554 int32_t ctdb_control_set_db_priority(struct ctdb_context *ctdb, TDB_DATA indata,
1555 uint32_t client_id)
1556 {
1557 struct ctdb_db_priority *db_prio = (struct ctdb_db_priority *)indata.dptr;
1558 struct ctdb_db_context *ctdb_db;
1559
1560 ctdb_db = find_ctdb_db(ctdb, db_prio->db_id);
1561 if (!ctdb_db) {
1562 if (!(ctdb->nodes[ctdb->pnn]->flags & NODE_FLAGS_INACTIVE)) {
1563 DEBUG(DEBUG_ERR,("Unknown db_id 0x%x in ctdb_set_db_priority\n",
1564 db_prio->db_id));
1565 }
1566 return 0;
1567 }
1568
1569 if ((db_prio->priority<1) || (db_prio->priority>NUM_DB_PRIORITIES)) {
1570 DEBUG(DEBUG_ERR,("Trying to set invalid priority : %u\n", db_prio->priority));
1571 return 0;
1572 }
1573
1574 ctdb_db->priority = db_prio->priority;
1575 DEBUG(DEBUG_INFO,("Setting DB priority to %u for db 0x%08x\n", db_prio->priority, db_prio->db_id));
1576
1577 if (client_id != 0) {
1578 /* Broadcast the update to the rest of the cluster */
1579 ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_ALL, 0,
1580 CTDB_CONTROL_SET_DB_PRIORITY, 0,
1581 CTDB_CTRL_FLAG_NOREPLY, indata,
1582 NULL, NULL);
1583 }
1584 return 0;
1585 }
1586
1587
1588 int ctdb_set_db_sticky(struct ctdb_context *ctdb, struct ctdb_db_context *ctdb_db)
1589 {
1590 if (ctdb_db->sticky) {
1591 return 0;
1592 }
1593
1594 if (ctdb_db->persistent) {
1595 DEBUG(DEBUG_ERR,("Trying to set persistent database with sticky property\n"));
1596 return -1;
1597 }
1598
1599 ctdb_db->sticky_records = trbt_create(ctdb_db, 0);
1600
1601 ctdb_db->sticky = true;
1602
1603 DEBUG(DEBUG_NOTICE,("set db sticky %s\n", ctdb_db->db_name));
1604
1605 return 0;
1606 }
1607
1608 int32_t ctdb_control_get_db_statistics(struct ctdb_context *ctdb,
1609 uint32_t db_id,
1610 TDB_DATA *outdata)
1611 {
1612 struct ctdb_db_context *ctdb_db;
1613 struct ctdb_db_statistics *stats;
1614 int i;
1615 int len;
1616 char *ptr;
1617
1618 ctdb_db = find_ctdb_db(ctdb, db_id);
1619 if (!ctdb_db) {
1620 DEBUG(DEBUG_ERR,("Unknown db_id 0x%x in get_db_statistics\n", db_id));
1621 return -1;
1622 }
1623
1624 len = offsetof(struct ctdb_db_statistics, hot_keys_wire);
1625 for (i = 0; i < MAX_HOT_KEYS; i++) {
1626 len += ctdb_db->statistics.hot_keys[i].key.dsize;
1627 }
1628
1629 stats = talloc_size(outdata, len);
1630 if (stats == NULL) {
1631 DEBUG(DEBUG_ERR,("Failed to allocate db statistics structure\n"));
1632 return -1;
1633 }
1634
1635 *stats = ctdb_db->statistics;
1636
1637 stats->num_hot_keys = MAX_HOT_KEYS;
1638
1639 ptr = &stats->hot_keys_wire[0];
1640 for (i = 0; i < MAX_HOT_KEYS; i++) {
1641 memcpy(ptr, ctdb_db->statistics.hot_keys[i].key.dptr,
1642 ctdb_db->statistics.hot_keys[i].key.dsize);
1643 ptr += ctdb_db->statistics.hot_keys[i].key.dsize;
1644 }
1645
1646 outdata->dptr = (uint8_t *)stats;
1647 outdata->dsize = len;
1648
1649 return 0;
1650 }
Samba GIT mirror