X-Git-Url: http://git.samba.org/?a=blobdiff_plain;f=tools%2Fctdb.c;h=674622a68afa129dabbf86a5a019f316964d4702;hb=bb1618faa065c3f245a4dc5c813e0e1c0ae9edde;hp=9d6547c55c8016da34cc714c6ba9fd484eedfc27;hpb=8c89aac20260dc7f3746e29fe99f17422a77cb88;p=sahlberg%2Fctdb.git

diff --git a/tools/ctdb.c b/tools/ctdb.c
index 9d6547c5..674622a6 100644
--- a/tools/ctdb.c
+++ b/tools/ctdb.c
@@ -19,7 +19,7 @@
 */
 
 #include "includes.h"
-#include "lib/events/events.h"
+#include "lib/tevent/tevent.h"
 #include "system/time.h"
 #include "system/filesys.h"
 #include "system/network.h"
@@ -27,6 +27,7 @@
 #include "popt.h"
 #include "cmdline.h"
 #include "../include/ctdb.h"
+#include "../include/ctdb_client.h"
 #include "../include/ctdb_private.h"
 #include "../common/rb_tree.h"
 #include "db_wrap.h"
@@ -35,16 +36,20 @@
 #define ERR_NONODE	21	/* node does not exist */
 #define ERR_DISNODE	22	/* node is disconnected */
 
+struct ctdb_connection *ctdb_connection;
+
 static void usage(void);
 
 static struct {
 	int timelimit;
 	uint32_t pnn;
 	int machinereadable;
+	int verbose;
 	int maxruntime;
 } options;
 
 #define TIMELIMIT() timeval_current_ofs(options.timelimit, 0)
+#define LONGTIMELIMIT() timeval_current_ofs(options.timelimit*10, 0)
 
 #ifdef CTDB_VERS
 static int control_version(struct ctdb_context *ctdb, int argc, const char **argv)
@@ -93,7 +98,7 @@ static void verify_node(struct ctdb_context *ctdb)
 	/* verify we can access the node */
 	ret = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);
 	if (ret == -1) {
-		DEBUG(DEBUG_ERR,("Can not ban node. Node is not operational.\n"));
+		DEBUG(DEBUG_ERR,("Can not access node. Node is not operational.\n"));
 		exit(10);
 	}
 }
@@ -152,12 +157,13 @@ static int control_process_exists(struct ctdb_context *ctdb, int argc, const cha
 /*
   display statistics structure
  */
-static void show_statistics(struct ctdb_statistics *s)
+static void show_statistics(struct ctdb_statistics *s, int show_header)
 {
 	TALLOC_CTX *tmp_ctx = talloc_new(NULL);
 	int i;
 	const char *prefix=NULL;
 	int preflen=0;
+	int tmp, days, hours, minutes, seconds;
 	const struct {
 		const char *name;
 		uint32_t offset;
@@ -166,6 +172,7 @@ static void show_statistics(struct ctdb_statistics *s)
 		STATISTICS_FIELD(num_clients),
 		STATISTICS_FIELD(frozen),
 		STATISTICS_FIELD(recovering),
+		STATISTICS_FIELD(num_recoveries),
 		STATISTICS_FIELD(client_packets_sent),
 		STATISTICS_FIELD(client_packets_recv),
 		STATISTICS_FIELD(node_packets_sent),
@@ -195,29 +202,110 @@ static void show_statistics(struct ctdb_statistics *s)
 		STATISTICS_FIELD(memory_used),
 		STATISTICS_FIELD(max_hop_count),
 	};
-	printf("CTDB version %u\n", CTDB_VERSION);
-	for (i=0;i<ARRAY_SIZE(fields);i++) {
-		if (strchr(fields[i].name, '.')) {
-			preflen = strcspn(fields[i].name, ".")+1;
-			if (!prefix || strncmp(prefix, fields[i].name, preflen) != 0) {
-				prefix = fields[i].name;
-				printf(" %*.*s\n", preflen-1, preflen-1, fields[i].name);
+	tmp = s->statistics_current_time.tv_sec - s->statistics_start_time.tv_sec;
+	seconds = tmp%60;
+	tmp    /= 60;
+	minutes = tmp%60;
+	tmp    /= 60;
+	hours   = tmp%24;
+	tmp    /= 24;
+	days    = tmp;
+
+	if (options.machinereadable){
+		if (show_header) {
+			printf("CTDB version:");
+			printf("Current time of statistics:");
+			printf("Statistics collected since:");
+			for (i=0;i<ARRAY_SIZE(fields);i++) {
+				printf("%s:", fields[i].name);
 			}
-		} else {
-			preflen = 0;
+			printf("num_reclock_ctdbd_latency:");
+			printf("min_reclock_ctdbd_latency:");
+			printf("avg_reclock_ctdbd_latency:");
+			printf("max_reclock_ctdbd_latency:");
+
+			printf("num_reclock_recd_latency:");
+			printf("min_reclock_recd_latency:");
+			printf("avg_reclock_recd_latency:");
+			printf("max_reclock_recd_latency:");
+
+			printf("num_call_latency:");
+			printf("min_call_latency:");
+			printf("avg_call_latency:");
+			printf("max_call_latency:");
+
+			printf("num_lockwait_latency:");
+			printf("min_lockwait_latency:");
+			printf("avg_lockwait_latency:");
+			printf("max_lockwait_latency:");
+
+			printf("num_childwrite_latency:");
+			printf("min_childwrite_latency:");
+			printf("avg_childwrite_latency:");
+			printf("max_childwrite_latency:");
+			printf("\n");
+		}
+		printf("%d:", CTDB_VERSION);
+		printf("%d:", (int)s->statistics_current_time.tv_sec);
+		printf("%d:", (int)s->statistics_start_time.tv_sec);
+		for (i=0;i<ARRAY_SIZE(fields);i++) {
+			printf("%d:", *(uint32_t *)(fields[i].offset+(uint8_t *)s));
+		}
+		printf("%d:", s->reclock.ctdbd.num);
+		printf("%.6f:", s->reclock.ctdbd.min);
+		printf("%.6f:", s->reclock.ctdbd.num?s->reclock.ctdbd.total/s->reclock.ctdbd.num:0.0);
+		printf("%.6f:", s->reclock.ctdbd.max);
+
+		printf("%d:", s->reclock.recd.num);
+		printf("%.6f:", s->reclock.recd.min);
+		printf("%.6f:", s->reclock.recd.num?s->reclock.recd.total/s->reclock.recd.num:0.0);
+		printf("%.6f:", s->reclock.recd.max);
+
+		printf("%d:", s->call_latency.num);
+		printf("%.6f:", s->call_latency.min);
+		printf("%.6f:", s->call_latency.num?s->call_latency.total/s->call_latency.num:0.0);
+		printf("%.6f:", s->call_latency.max);
+
+		printf("%d:", s->lockwait_latency.num);
+		printf("%.6f:", s->lockwait_latency.min);
+		printf("%.6f:", s->lockwait_latency.num?s->lockwait_latency.total/s->lockwait_latency.num:0.0);
+		printf("%.6f:", s->lockwait_latency.max);
+
+		printf("%d:", s->childwrite_latency.num);
+		printf("%.6f:", s->childwrite_latency.min);
+		printf("%.6f:", s->childwrite_latency.num?s->childwrite_latency.total/s->childwrite_latency.num:0.0);
+		printf("%.6f:", s->childwrite_latency.max);
+		printf("\n");
+	} else {
+		printf("CTDB version %u\n", CTDB_VERSION);
+		printf("Current time of statistics  :                %s", ctime(&s->statistics_current_time.tv_sec));
+		printf("Statistics collected since  : (%03d %02d:%02d:%02d) %s", days, hours, minutes, seconds, ctime(&s->statistics_start_time.tv_sec));
+
+		for (i=0;i<ARRAY_SIZE(fields);i++) {
+			if (strchr(fields[i].name, '.')) {
+				preflen = strcspn(fields[i].name, ".")+1;
+				if (!prefix || strncmp(prefix, fields[i].name, preflen) != 0) {
+					prefix = fields[i].name;
+					printf(" %*.*s\n", preflen-1, preflen-1, fields[i].name);
+				}
+			} else {
+				preflen = 0;
+			}
+			printf(" %*s%-22s%*s%10u\n", 
+			       preflen?4:0, "",
+			       fields[i].name+preflen, 
+			       preflen?0:4, "",
+			       *(uint32_t *)(fields[i].offset+(uint8_t *)s));
 		}
-		printf(" %*s%-22s%*s%10u\n", 
-		       preflen?4:0, "",
-		       fields[i].name+preflen, 
-		       preflen?0:4, "",
-		       *(uint32_t *)(fields[i].offset+(uint8_t *)s));
+		printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n", "reclock_ctdbd       MIN/AVG/MAX", s->reclock.ctdbd.min, s->reclock.ctdbd.num?s->reclock.ctdbd.total/s->reclock.ctdbd.num:0.0, s->reclock.ctdbd.max, s->reclock.ctdbd.num);
+
+		printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n", "reclock_recd       MIN/AVG/MAX", s->reclock.recd.min, s->reclock.recd.num?s->reclock.recd.total/s->reclock.recd.num:0.0, s->reclock.recd.max, s->reclock.recd.num);
+
+		printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n", "call_latency       MIN/AVG/MAX", s->call_latency.min, s->call_latency.num?s->call_latency.total/s->call_latency.num:0.0, s->call_latency.max, s->call_latency.num);
+		printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n", "lockwait_latency   MIN/AVG/MAX", s->lockwait_latency.min, s->lockwait_latency.num?s->lockwait_latency.total/s->lockwait_latency.num:0.0, s->lockwait_latency.max, s->lockwait_latency.num);
+		printf(" %-30s     %.6f/%.6f/%.6f sec out of %d\n", "childwrite_latency MIN/AVG/MAX", s->childwrite_latency.min, s->childwrite_latency.num?s->childwrite_latency.total/s->childwrite_latency.num:0.0, s->childwrite_latency.max, s->childwrite_latency.num);
 	}
-	printf(" %-30s     %.6f sec\n", "max_reclock_ctdbd", s->reclock.ctdbd);
-	printf(" %-30s     %.6f sec\n", "max_reclock_recd", s->reclock.recd);
 
-	printf(" %-30s     %.6f sec\n", "max_call_latency", s->max_call_latency);
-	printf(" %-30s     %.6f sec\n", "max_lockwait_latency", s->max_lockwait_latency);
-	printf(" %-30s     %.6f sec\n", "max_childwrite_latency", s->max_childwrite_latency);
 	talloc_free(tmp_ctx);
 }
 
@@ -253,14 +341,14 @@ static int control_statistics_all(struct ctdb_context *ctdb)
 		}
 		statistics.max_hop_count = 
 			MAX(statistics.max_hop_count, s1.max_hop_count);
-		statistics.max_call_latency = 
-			MAX(statistics.max_call_latency, s1.max_call_latency);
-		statistics.max_lockwait_latency = 
-			MAX(statistics.max_lockwait_latency, s1.max_lockwait_latency);
+		statistics.call_latency.max = 
+			MAX(statistics.call_latency.max, s1.call_latency.max);
+		statistics.lockwait_latency.max = 
+			MAX(statistics.lockwait_latency.max, s1.lockwait_latency.max);
 	}
 	talloc_free(nodes);
 	printf("Gathered statistics for %u nodes\n", num_nodes);
-	show_statistics(&statistics);
+	show_statistics(&statistics, 1);
 	return 0;
 }
 
@@ -281,7 +369,7 @@ static int control_statistics(struct ctdb_context *ctdb, int argc, const char **
 		DEBUG(DEBUG_ERR, ("Unable to get statistics from node %u\n", options.pnn));
 		return ret;
 	}
-	show_statistics(&statistics);
+	show_statistics(&statistics, 1);
 	return 0;
 }
 
@@ -302,6 +390,37 @@ static int control_statistics_reset(struct ctdb_context *ctdb, int argc, const c
 }
 
 
+/*
+  display remote ctdb rolling statistics
+ */
+static int control_stats(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	int ret;
+	struct ctdb_statistics_wire *stats;
+	int i, num_records = -1;
+
+	if (argc ==1) {
+		num_records = atoi(argv[0]) - 1;
+	}
+
+	ret = ctdb_ctrl_getstathistory(ctdb, TIMELIMIT(), options.pnn, ctdb, &stats);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to get rolling statistics from node %u\n", options.pnn));
+		return ret;
+	}
+	for (i=0;i<stats->num;i++) {
+		if (stats->stats[i].statistics_start_time.tv_sec == 0) {
+			continue;
+		}
+		show_statistics(&stats->stats[i], i==0);
+		if (i == num_records) {
+			break;
+		}
+	}
+	return 0;
+}
+
+
 /*
   display uptime of remote node
  */
@@ -363,11 +482,12 @@ static int control_uptime(struct ctdb_context *ctdb, int argc, const char **argv
  */
 static int control_pnn(struct ctdb_context *ctdb, int argc, const char **argv)
 {
-	int mypnn;
+	uint32_t mypnn;
+	bool ret;
 
-	mypnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), options.pnn);
-	if (mypnn == -1) {
-		DEBUG(DEBUG_ERR, ("Unable to get pnn from local node."));
+	ret = ctdb_getpnn(ctdb_connection, options.pnn, &mypnn);
+	if (!ret) {
+		DEBUG(DEBUG_ERR, ("Unable to get pnn from node."));
 		return -1;
 	}
 
@@ -502,40 +622,19 @@ static int control_status(struct ctdb_context *ctdb, int argc, const char **argv
 	}
 
 	if(options.machinereadable){
-		printf(":Node:IP:Disconnected:Banned:Disabled:Unhealthy:Stopped:Inactive:PartiallyOnline:\n");
+		printf(":Node:IP:Disconnected:Banned:Disabled:Unhealthy:Stopped:Inactive:\n");
 		for(i=0;i<nodemap->num;i++){
-			int partially_online = 0;
-			int j;
-
 			if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
 				continue;
 			}
-			if (nodemap->nodes[i].flags == 0) {
-				struct ctdb_control_get_ifaces *ifaces;
-
-				ret = ctdb_ctrl_get_ifaces(ctdb, TIMELIMIT(),
-							   nodemap->nodes[i].pnn,
-							   ctdb, &ifaces);
-				if (ret == 0) {
-					for (j=0; j < ifaces->num; j++) {
-						if (ifaces->ifaces[j].link_state != 0) {
-							continue;
-						}
-						partially_online = 1;
-						break;
-					}
-					talloc_free(ifaces);
-				}
-			}
-			printf(":%d:%s:%d:%d:%d:%d:%d:%d:%d:\n", nodemap->nodes[i].pnn,
+			printf(":%d:%s:%d:%d:%d:%d:%d:%d:\n", nodemap->nodes[i].pnn,
 				ctdb_addr_to_str(&nodemap->nodes[i].addr),
 			       !!(nodemap->nodes[i].flags&NODE_FLAGS_DISCONNECTED),
 			       !!(nodemap->nodes[i].flags&NODE_FLAGS_BANNED),
 			       !!(nodemap->nodes[i].flags&NODE_FLAGS_PERMANENTLY_DISABLED),
 			       !!(nodemap->nodes[i].flags&NODE_FLAGS_UNHEALTHY),
 			       !!(nodemap->nodes[i].flags&NODE_FLAGS_STOPPED),
-			       !!(nodemap->nodes[i].flags&NODE_FLAGS_INACTIVE),
-			       partially_online);
+			       !!(nodemap->nodes[i].flags&NODE_FLAGS_INACTIVE));
 		}
 		return 0;
 	}
@@ -560,23 +659,6 @@ static int control_status(struct ctdb_context *ctdb, int argc, const char **argv
 		if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
 			continue;
 		}
-		if (nodemap->nodes[i].flags == 0) {
-			struct ctdb_control_get_ifaces *ifaces;
-
-			ret = ctdb_ctrl_get_ifaces(ctdb, TIMELIMIT(),
-						   nodemap->nodes[i].pnn,
-						   ctdb, &ifaces);
-			if (ret == 0) {
-				for (j=0; j < ifaces->num; j++) {
-					if (ifaces->ifaces[j].link_state != 0) {
-						continue;
-					}
-					flags_str = talloc_strdup(ctdb, "PARTIALLYONLINE");
-					break;
-				}
-				talloc_free(ifaces);
-			}
-		}
 		for (j=0;j<ARRAY_SIZE(flag_names);j++) {
 			if (nodemap->nodes[i].flags & flag_names[j].flag) {
 				if (flags_str == NULL) {
@@ -643,6 +725,7 @@ struct natgw_node {
 static int control_natgwlist(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	int i, ret;
+	uint32_t capabilities;
 	const char *natgw_list;
 	int nlines;
 	char **lines;
@@ -721,6 +804,14 @@ static int control_natgwlist(struct ctdb_context *ctdb, int argc, const char **a
 	 */
 	for(i=0;i<nodemap->num;i++){
 		if (!(nodemap->nodes[i].flags & (NODE_FLAGS_DISCONNECTED|NODE_FLAGS_STOPPED|NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_UNHEALTHY))) {
+			ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, &capabilities);
+			if (ret != 0) {
+				DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", nodemap->nodes[i].pnn));
+				return ret;
+			}
+			if (!(capabilities&CTDB_CAP_NATGW)) {
+				continue;
+			}
 			printf("%d %s\n", nodemap->nodes[i].pnn,ctdb_addr_to_str(&nodemap->nodes[i].addr));
 			break;
 		}
@@ -729,6 +820,14 @@ static int control_natgwlist(struct ctdb_context *ctdb, int argc, const char **a
 	if (i == nodemap->num) {
 		for(i=0;i<nodemap->num;i++){
 			if (!(nodemap->nodes[i].flags & (NODE_FLAGS_DISCONNECTED|NODE_FLAGS_STOPPED|NODE_FLAGS_DELETED))) {
+				ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, &capabilities);
+				if (ret != 0) {
+					DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", nodemap->nodes[i].pnn));
+					return ret;
+				}
+				if (!(capabilities&CTDB_CAP_NATGW)) {
+					continue;
+				}
 				printf("%d %s\n", nodemap->nodes[i].pnn,ctdb_addr_to_str(&nodemap->nodes[i].addr));
 				break;
 			}
@@ -738,6 +837,14 @@ static int control_natgwlist(struct ctdb_context *ctdb, int argc, const char **a
 	if (i == nodemap->num) {
 		for(i=0;i<nodemap->num;i++){
 			if (!(nodemap->nodes[i].flags & (NODE_FLAGS_DISCONNECTED|NODE_FLAGS_DELETED))) {
+				ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, &capabilities);
+				if (ret != 0) {
+					DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", nodemap->nodes[i].pnn));
+					return ret;
+				}
+				if (!(capabilities&CTDB_CAP_NATGW)) {
+					continue;
+				}
 				printf("%d %s\n", nodemap->nodes[i].pnn, ctdb_addr_to_str(&nodemap->nodes[i].addr));
 				break;
 			}
@@ -959,6 +1066,80 @@ static int control_recmaster(struct ctdb_context *ctdb, int argc, const char **a
 	return 0;
 }
 
+/*
+  add a tickle to a public address
+ */
+static int control_add_tickle(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	struct ctdb_tcp_connection t;
+	TDB_DATA data;
+	int ret;
+
+	if (argc < 2) {
+		usage();
+	}
+
+	if (parse_ip_port(argv[0], &t.src_addr) == 0) {
+		DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
+		return -1;
+	}
+	if (parse_ip_port(argv[1], &t.dst_addr) == 0) {
+		DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[1]));
+		return -1;
+	}
+
+	data.dptr = (uint8_t *)&t;
+	data.dsize = sizeof(t);
+
+	/* tell all nodes about this tcp connection */
+	ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_TCP_ADD_DELAYED_UPDATE,
+			   0, data, ctdb, NULL, NULL, NULL, NULL);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to add tickle\n"));
+		return -1;
+	}
+	
+	return 0;
+}
+
+
+/*
+  delete a tickle from a node
+ */
+static int control_del_tickle(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	struct ctdb_tcp_connection t;
+	TDB_DATA data;
+	int ret;
+
+	if (argc < 2) {
+		usage();
+	}
+
+	if (parse_ip_port(argv[0], &t.src_addr) == 0) {
+		DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
+		return -1;
+	}
+	if (parse_ip_port(argv[1], &t.dst_addr) == 0) {
+		DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[1]));
+		return -1;
+	}
+
+	data.dptr = (uint8_t *)&t;
+	data.dsize = sizeof(t);
+
+	/* tell all nodes about this tcp connection */
+	ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_TCP_REMOVE,
+			   0, data, ctdb, NULL, NULL, NULL, NULL);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to remove tickle\n"));
+		return -1;
+	}
+	
+	return 0;
+}
+
+
 /*
   get a list of all tickles for this pnn
  */
@@ -967,11 +1148,16 @@ static int control_get_tickles(struct ctdb_context *ctdb, int argc, const char *
 	struct ctdb_control_tcp_tickle_list *list;
 	ctdb_sock_addr addr;
 	int i, ret;
+	unsigned port = 0;
 
 	if (argc < 1) {
 		usage();
 	}
 
+	if (argc == 2) {
+		port = atoi(argv[1]);
+	}
+
 	if (parse_ip(argv[0], NULL, 0, &addr) == 0) {
 		DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s'\n", argv[0]));
 		return -1;
@@ -983,11 +1169,25 @@ static int control_get_tickles(struct ctdb_context *ctdb, int argc, const char *
 		return -1;
 	}
 
-	printf("Tickles for ip:%s\n", ctdb_addr_to_str(&list->addr));
-	printf("Num tickles:%u\n", list->tickles.num);
-	for (i=0;i<list->tickles.num;i++) {
-		printf("SRC: %s:%u   ", ctdb_addr_to_str(&list->tickles.connections[i].src_addr), ntohs(list->tickles.connections[i].src_addr.ip.sin_port));
-		printf("DST: %s:%u\n", ctdb_addr_to_str(&list->tickles.connections[i].dst_addr), ntohs(list->tickles.connections[i].dst_addr.ip.sin_port));
+	if (options.machinereadable){
+		printf(":source ip:port:destination ip:port:\n");
+		for (i=0;i<list->tickles.num;i++) {
+			if (port && port != ntohs(list->tickles.connections[i].dst_addr.ip.sin_port)) {
+				continue;
+			}
+			printf(":%s:%u", ctdb_addr_to_str(&list->tickles.connections[i].src_addr), ntohs(list->tickles.connections[i].src_addr.ip.sin_port));
+			printf(":%s:%u:\n", ctdb_addr_to_str(&list->tickles.connections[i].dst_addr), ntohs(list->tickles.connections[i].dst_addr.ip.sin_port));
+		}
+	} else {
+		printf("Tickles for ip:%s\n", ctdb_addr_to_str(&list->addr));
+		printf("Num tickles:%u\n", list->tickles.num);
+		for (i=0;i<list->tickles.num;i++) {
+			if (port && port != ntohs(list->tickles.connections[i].dst_addr.ip.sin_port)) {
+				continue;
+			}
+			printf("SRC: %s:%u   ", ctdb_addr_to_str(&list->tickles.connections[i].src_addr), ntohs(list->tickles.connections[i].src_addr.ip.sin_port));
+			printf("DST: %s:%u\n", ctdb_addr_to_str(&list->tickles.connections[i].dst_addr), ntohs(list->tickles.connections[i].dst_addr.ip.sin_port));
+		}
 	}
 
 	talloc_free(list);
@@ -996,7 +1196,6 @@ static int control_get_tickles(struct ctdb_context *ctdb, int argc, const char *
 }
 
 
-
 static int move_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr, uint32_t pnn)
 {
 	struct ctdb_all_public_ips *ips;
@@ -1011,7 +1210,7 @@ static int move_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr, uint32_t pnn
 	disable_time = 30;
 	data.dptr  = (uint8_t*)&disable_time;
 	data.dsize = sizeof(disable_time);
-	ret = ctdb_send_message(ctdb, CTDB_BROADCAST_CONNECTED, CTDB_SRVID_DISABLE_IP_CHECK, data);
+	ret = ctdb_client_send_message(ctdb, CTDB_BROADCAST_CONNECTED, CTDB_SRVID_DISABLE_IP_CHECK, data);
 	if (ret != 0) {
 		DEBUG(DEBUG_ERR,("Failed to send message to disable ipcheck\n"));
 		return -1;
@@ -1055,7 +1254,7 @@ static int move_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr, uint32_t pnn
        	nodes = list_of_active_nodes_except_pnn(ctdb, nodemap, tmp_ctx, pnn);
 	ret = ctdb_client_async_control(ctdb, CTDB_CONTROL_RELEASE_IP,
 					nodes, 0,
-					TIMELIMIT(),
+					LONGTIMELIMIT(),
 					false, data,
 					NULL, NULL,
 					NULL);
@@ -1065,13 +1264,22 @@ static int move_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr, uint32_t pnn
 		return -1;
 	}
 
-	ret = ctdb_ctrl_takeover_ip(ctdb, TIMELIMIT(), pnn, &ip);
+	ret = ctdb_ctrl_takeover_ip(ctdb, LONGTIMELIMIT(), pnn, &ip);
 	if (ret != 0) {
 		DEBUG(DEBUG_ERR,("Failed to take over IP on node %d\n", pnn));
 		talloc_free(tmp_ctx);
 		return -1;
 	}
 
+	/* update the recovery daemon so it now knows to expect the new
+	   node assignment for this ip.
+	*/
+	ret = ctdb_client_send_message(ctdb, CTDB_BROADCAST_CONNECTED, CTDB_SRVID_RECD_UPDATE_IP, data);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to send message to update the ip on the recovery master.\n"));
+		return -1;
+	}
+
 	talloc_free(tmp_ctx);
 	return 0;
 }
@@ -1082,6 +1290,7 @@ static int move_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr, uint32_t pnn
 static int control_moveip(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	uint32_t pnn;
+	int ret, retries = 0;
 	ctdb_sock_addr addr;
 
 	if (argc < 2) {
@@ -1100,8 +1309,16 @@ static int control_moveip(struct ctdb_context *ctdb, int argc, const char **argv
 		return -1;
 	}
 
-	if (move_ip(ctdb, &addr, pnn) != 0) {
-		DEBUG(DEBUG_ERR,("Failed to move ip to node %d\n", pnn));
+	do {
+		ret = move_ip(ctdb, &addr, pnn);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR,("Failed to move ip to node %d. Wait 3 second and try again.\n", pnn));
+			sleep(3);
+			retries++;
+		}
+	} while (retries < 5 && ret != 0);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to move ip to node %d. Giving up.\n", pnn));
 		return -1;
 	}
 
@@ -1265,103 +1482,253 @@ find_other_host_for_public_ip(struct ctdb_context *ctdb, ctdb_sock_addr *addr)
 	return -1;
 }
 
+static uint32_t ipreallocate_finished;
+
 /*
-  add a public ip address to a node
+  handler for receiving the response to ipreallocate
+*/
+static void ip_reallocate_handler(struct ctdb_context *ctdb, uint64_t srvid, 
+			     TDB_DATA data, void *private_data)
+{
+	ipreallocate_finished = 1;
+}
+
+static void ctdb_every_second(struct event_context *ev, struct timed_event *te, struct timeval t, void *p)
+{
+	struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
+
+	event_add_timed(ctdb->ev, ctdb, 
+				timeval_current_ofs(1, 0),
+				ctdb_every_second, ctdb);
+}
+
+/*
+  ask the recovery daemon on the recovery master to perform a ip reallocation
  */
-static int control_addip(struct ctdb_context *ctdb, int argc, const char **argv)
+static int control_ipreallocate(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	int i, ret;
-	int len;
-	uint32_t pnn;
-	unsigned mask;
-	ctdb_sock_addr addr;
-	struct ctdb_control_ip_iface *pub;
-	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
-	struct ctdb_all_public_ips *ips;
-
+	TDB_DATA data;
+	struct takeover_run_reply rd;
+	uint32_t recmaster;
+	struct ctdb_node_map *nodemap=NULL;
+	int retries=0;
+	struct timeval tv = timeval_current();
 
-	if (argc != 2) {
-		talloc_free(tmp_ctx);
-		usage();
-	}
+	/* we need some events to trigger so we can timeout and restart
+	   the loop
+	*/
+	event_add_timed(ctdb->ev, ctdb, 
+				timeval_current_ofs(1, 0),
+				ctdb_every_second, ctdb);
 
-	if (!parse_ip_mask(argv[0], argv[1], &addr, &mask)) {
-		DEBUG(DEBUG_ERR, ("Badly formed ip/mask : %s\n", argv[0]));
-		talloc_free(tmp_ctx);
+	rd.pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
+	if (rd.pnn == -1) {
+		DEBUG(DEBUG_ERR, ("Failed to get pnn of local node\n"));
 		return -1;
 	}
+	rd.srvid = getpid();
 
-	ret = control_get_all_public_ips(ctdb, tmp_ctx, &ips);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to get public ip list from cluster\n"));
-		talloc_free(tmp_ctx);
-		return ret;
-	}
+	/* register a message port for receiveing the reply so that we
+	   can receive the reply
+	*/
+	ctdb_client_set_message_handler(ctdb, rd.srvid, ip_reallocate_handler, NULL);
 
+	data.dptr = (uint8_t *)&rd;
+	data.dsize = sizeof(rd);
 
-	/* check if some other node is already serving this ip, if not,
-	 * we will claim it
-	 */
-	for (i=0;i<ips->num;i++) {
-		if (ctdb_same_ip(&addr, &ips->ips[i].addr)) {
+again:
+	/* check that there are valid nodes available */
+	if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap) != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
+		return -1;
+	}
+	for (i=0; i<nodemap->num;i++) {
+		if ((nodemap->nodes[i].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) == 0) {
 			break;
 		}
 	}
+	if (i==nodemap->num) {
+		DEBUG(DEBUG_ERR,("No recmaster available, no need to wait for cluster convergence\n"));
+		return 0;
+	}
 
-	len = offsetof(struct ctdb_control_ip_iface, iface) + strlen(argv[1]) + 1;
-	pub = talloc_size(tmp_ctx, len); 
-	CTDB_NO_MEMORY(ctdb, pub);
-
-	pub->addr  = addr;
-	pub->mask  = mask;
-	pub->len   = strlen(argv[1])+1;
-	memcpy(&pub->iface[0], argv[1], strlen(argv[1])+1);
 
-	ret = ctdb_ctrl_add_public_ip(ctdb, TIMELIMIT(), options.pnn, pub);
+	ret = ctdb_ctrl_getrecmaster(ctdb, ctdb, TIMELIMIT(), options.pnn, &recmaster);
 	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to add public ip to node %u\n", options.pnn));
-		talloc_free(tmp_ctx);
+		DEBUG(DEBUG_ERR, ("Unable to get recmaster from node %u\n", options.pnn));
 		return ret;
 	}
 
-	if (i == ips->num) {
-		/* no one has this ip so we claim it */
-		pnn  = options.pnn;
-	} else {
-		pnn  = ips->ips[i].pnn;
-	}
-
-	if (move_ip(ctdb, &addr, pnn) != 0) {
-		DEBUG(DEBUG_ERR,("Failed to move ip to node %d\n", pnn));
+	/* verify the node exists */
+	if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), recmaster, ctdb, &nodemap) != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
 		return -1;
 	}
 
-	talloc_free(tmp_ctx);
-	return 0;
-}
-
-static int control_delip(struct ctdb_context *ctdb, int argc, const char **argv);
-
-static int control_delip_all(struct ctdb_context *ctdb, int argc, const char **argv, ctdb_sock_addr *addr)
-{
-	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
-	struct ctdb_node_map *nodemap=NULL;
-	struct ctdb_all_public_ips *ips;
-	int ret, i, j;
-
-	ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, tmp_ctx, &nodemap);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to get nodemap from current node\n"));
-		return ret;
-	}
 
-	/* remove it from the nodes that are not hosting the ip currently */
-	for(i=0;i<nodemap->num;i++){
-		if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
-			continue;
-		}
-		if (ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, tmp_ctx, &ips) != 0) {
-			DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %d\n", nodemap->nodes[i].pnn));
+	/* check tha there are nodes available that can act as a recmaster */
+	for (i=0; i<nodemap->num; i++) {
+		if (nodemap->nodes[i].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
+			continue;
+		}
+		break;
+	}
+	if (i == nodemap->num) {
+		DEBUG(DEBUG_ERR,("No possible nodes to host addresses.\n"));
+		return 0;
+	}
+
+	/* verify the recovery master is not STOPPED, nor BANNED */
+	if (nodemap->nodes[recmaster].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
+		DEBUG(DEBUG_ERR,("No suitable recmaster found. Try again\n"));
+		retries++;
+		sleep(1);
+		goto again;
+	} 
+	
+	/* verify the recovery master is not STOPPED, nor BANNED */
+	if (nodemap->nodes[recmaster].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
+		DEBUG(DEBUG_ERR,("No suitable recmaster found. Try again\n"));
+		retries++;
+		sleep(1);
+		goto again;
+	} 
+
+	ipreallocate_finished = 0;
+	ret = ctdb_client_send_message(ctdb, recmaster, CTDB_SRVID_TAKEOVER_RUN, data);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to send ip takeover run request message to %u\n", options.pnn));
+		return -1;
+	}
+
+	tv = timeval_current();
+	/* this loop will terminate when we have received the reply */
+	while (timeval_elapsed(&tv) < 5.0 && ipreallocate_finished == 0) {
+		event_loop_once(ctdb->ev);
+	}
+	if (ipreallocate_finished == 1) {
+		return 0;
+	}
+
+	retries++;
+	sleep(1);
+	goto again;
+
+	return 0;
+}
+
+
+/*
+  add a public ip address to a node
+ */
+static int control_addip(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	int i, ret;
+	int len, retries = 0;
+	unsigned mask;
+	ctdb_sock_addr addr;
+	struct ctdb_control_ip_iface *pub;
+	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
+	struct ctdb_all_public_ips *ips;
+
+
+	if (argc != 2) {
+		talloc_free(tmp_ctx);
+		usage();
+	}
+
+	if (!parse_ip_mask(argv[0], argv[1], &addr, &mask)) {
+		DEBUG(DEBUG_ERR, ("Badly formed ip/mask : %s\n", argv[0]));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
+
+	/* read the public ip list from the node */
+	ret = ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), options.pnn, tmp_ctx, &ips);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %u\n", options.pnn));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
+	for (i=0;i<ips->num;i++) {
+		if (ctdb_same_ip(&addr, &ips->ips[i].addr)) {
+			DEBUG(DEBUG_ERR,("Can not add ip to node. Node already hosts this ip\n"));
+			return 0;
+		}
+	}
+
+
+
+	/* Dont timeout. This command waits for an ip reallocation
+	   which sometimes can take wuite a while if there has
+	   been a recent recovery
+	*/
+	alarm(0);
+
+	len = offsetof(struct ctdb_control_ip_iface, iface) + strlen(argv[1]) + 1;
+	pub = talloc_size(tmp_ctx, len); 
+	CTDB_NO_MEMORY(ctdb, pub);
+
+	pub->addr  = addr;
+	pub->mask  = mask;
+	pub->len   = strlen(argv[1])+1;
+	memcpy(&pub->iface[0], argv[1], strlen(argv[1])+1);
+
+	do {
+		ret = ctdb_ctrl_add_public_ip(ctdb, TIMELIMIT(), options.pnn, pub);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, ("Unable to add public ip to node %u. Wait 3 seconds and try again.\n", options.pnn));
+			sleep(3);
+			retries++;
+		}
+	} while (retries < 5 && ret != 0);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to add public ip to node %u. Giving up.\n", options.pnn));
+		talloc_free(tmp_ctx);
+		return ret;
+	}
+
+	do {
+		ret = control_ipreallocate(ctdb, argc, argv);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u. Wait 3 seconds and try again.\n", options.pnn));
+			sleep(3);
+			retries++;
+		}
+	} while (retries < 5 && ret != 0);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u. Giving up.\n", options.pnn));
+		talloc_free(tmp_ctx);
+		return ret;
+	}
+
+	talloc_free(tmp_ctx);
+	return 0;
+}
+
+static int control_delip(struct ctdb_context *ctdb, int argc, const char **argv);
+
+static int control_delip_all(struct ctdb_context *ctdb, int argc, const char **argv, ctdb_sock_addr *addr)
+{
+	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
+	struct ctdb_node_map *nodemap=NULL;
+	struct ctdb_all_public_ips *ips;
+	int ret, i, j;
+
+	ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, tmp_ctx, &nodemap);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to get nodemap from current node\n"));
+		return ret;
+	}
+
+	/* remove it from the nodes that are not hosting the ip currently */
+	for(i=0;i<nodemap->num;i++){
+		if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
+			continue;
+		}
+		if (ctdb_ctrl_get_public_ips(ctdb, TIMELIMIT(), nodemap->nodes[i].pnn, tmp_ctx, &ips) != 0) {
+			DEBUG(DEBUG_ERR, ("Unable to get public ip list from node %d\n", nodemap->nodes[i].pnn));
 			continue;
 		}
 
@@ -1416,6 +1783,7 @@ static int control_delip_all(struct ctdb_context *ctdb, int argc, const char **a
 static int control_delip(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	int i, ret;
+	int retries = 0;
 	ctdb_sock_addr addr;
 	struct ctdb_control_ip_iface pub;
 	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
@@ -1462,8 +1830,16 @@ static int control_delip(struct ctdb_context *ctdb, int argc, const char **argv)
 	if (ips->ips[i].pnn == options.pnn) {
 		ret = find_other_host_for_public_ip(ctdb, &addr);
 		if (ret != -1) {
-			if (move_ip(ctdb, &addr, ret) != 0) {
-				DEBUG(DEBUG_ERR,("Failed to move ip to node %d\n", ret));
+			do {
+				ret = move_ip(ctdb, &addr, ret);
+				if (ret != 0) {
+					DEBUG(DEBUG_ERR,("Failed to move ip to node %d. Wait 3 seconds and try again.\n", options.pnn));
+					sleep(3);
+					retries++;
+				}
+			} while (retries < 5 && ret != 0);
+			if (ret != 0) {
+				DEBUG(DEBUG_ERR,("Failed to move ip to node %d. Giving up.\n", options.pnn));
 				return -1;
 			}
 		}
@@ -1559,6 +1935,8 @@ static int regsrvid(struct ctdb_context *ctdb, int argc, const char **argv)
 		DEBUG(DEBUG_ERR, ("Unable to register server_id from node %u\n", options.pnn));
 		return ret;
 	}
+	DEBUG(DEBUG_ERR,("Srvid registered. Sleeping for 999 seconds\n"));
+	sleep(999);
 	return -1;
 }
 
@@ -1696,7 +2074,11 @@ static int control_ip(struct ctdb_context *ctdb, int argc, const char **argv)
 	}
 
 	if (options.machinereadable){
-		printf(":Public IP:Node:ActiveInterface:AvailableInterfaces:ConfiguredInterfaces:\n");
+		printf(":Public IP:Node:");
+		if (options.verbose){
+			printf("ActiveInterface:AvailableInterfaces:ConfiguredInterfaces:");
+		}
+		printf("\n");
 	} else {
 		if (options.pnn == CTDB_BROADCAST_ALL) {
 			printf("Public IPs on ALL nodes\n");
@@ -1756,19 +2138,29 @@ static int control_ip(struct ctdb_context *ctdb, int argc, const char **argv)
 		}
 
 		if (options.machinereadable){
-			printf(":%s:%d:%s:%s:%s:\n",
-			       ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
-			       ips->ips[ips->num-i].pnn,
-			       aciface?aciface:"",
-			       avifaces?avifaces:"",
-			       cifaces?cifaces:"");
+			printf(":%s:%d:",
+				ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
+				ips->ips[ips->num-i].pnn);
+			if (options.verbose){
+				printf("%s:%s:%s:",
+					aciface?aciface:"",
+					avifaces?avifaces:"",
+					cifaces?cifaces:"");
+			}
+			printf("\n");
 		} else {
-			printf("%s node[%d] active[%s] available[%s] configured[%s]\n",
-			       ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
-			       ips->ips[ips->num-i].pnn,
-			       aciface?aciface:"",
-			       avifaces?avifaces:"",
-			       cifaces?cifaces:"");
+			if (options.verbose) {
+				printf("%s node[%d] active[%s] available[%s] configured[%s]\n",
+					ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
+					ips->ips[ips->num-i].pnn,
+					aciface?aciface:"",
+					avifaces?avifaces:"",
+					cifaces?cifaces:"");
+			} else {
+				printf("%s %d\n",
+					ctdb_addr_to_str(&ips->ips[ips->num-i].addr),
+					ips->ips[ips->num-i].pnn);
+			}
 		}
 		talloc_free(info);
 	}
@@ -1940,204 +2332,88 @@ static int control_getpid(struct ctdb_context *ctdb, int argc, const char **argv
 }
 
 /*
-  handler for receiving the response to ipreallocate
-*/
-static void ip_reallocate_handler(struct ctdb_context *ctdb, uint64_t srvid, 
-			     TDB_DATA data, void *private_data)
+  disable a remote node
+ */
+static int control_disable(struct ctdb_context *ctdb, int argc, const char **argv)
 {
-	exit(0);
-}
+	int ret;
+	struct ctdb_node_map *nodemap=NULL;
 
-static void ctdb_every_second(struct event_context *ev, struct timed_event *te, struct timeval t, void *p)
-{
-	struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
+	/* check if the node is already disabled */
+	if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
+		exit(10);
+	}
+	if (nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
+		DEBUG(DEBUG_ERR,("Node %d is already disabled.\n", options.pnn));
+		return 0;
+	}
 
-	event_add_timed(ctdb->ev, ctdb, 
-				timeval_current_ofs(1, 0),
-				ctdb_every_second, ctdb);
-}
+	do {
+		ret = ctdb_ctrl_modflags(ctdb, TIMELIMIT(), options.pnn, NODE_FLAGS_PERMANENTLY_DISABLED, 0);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, ("Unable to disable node %u\n", options.pnn));
+			return ret;
+		}
 
-/*
-  ask the recovery daemon on the recovery master to perform a ip reallocation
- */
-static int control_ipreallocate(struct ctdb_context *ctdb, int argc, const char **argv)
-{
-	int i, ret;
-	TDB_DATA data;
-	struct takeover_run_reply rd;
-	uint32_t recmaster;
-	struct ctdb_node_map *nodemap=NULL;
-	int retries=0;
-	struct timeval tv = timeval_current();
+		sleep(1);
 
-	/* we need some events to trigger so we can timeout and restart
-	   the loop
-	*/
-	event_add_timed(ctdb->ev, ctdb, 
-				timeval_current_ofs(1, 0),
-				ctdb_every_second, ctdb);
+		/* read the nodemap and verify the change took effect */
+		if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
+			DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
+			exit(10);
+		}
 
-	rd.pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
-	if (rd.pnn == -1) {
-		DEBUG(DEBUG_ERR, ("Failed to get pnn of local node\n"));
-		return -1;
+	} while (!(nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED));
+	ret = control_ipreallocate(ctdb, argc, argv);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
+		return ret;
 	}
-	rd.srvid = getpid();
 
-	/* register a message port for receiveing the reply so that we
-	   can receive the reply
-	*/
-	ctdb_set_message_handler(ctdb, rd.srvid, ip_reallocate_handler, NULL);
+	return 0;
+}
 
-	data.dptr = (uint8_t *)&rd;
-	data.dsize = sizeof(rd);
+/*
+  enable a disabled remote node
+ */
+static int control_enable(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	int ret;
 
-again:
-	if (retries>5) {
-		DEBUG(DEBUG_ERR,("Failed waiting for cluster convergense\n"));
-		exit(10);
-	}
+	struct ctdb_node_map *nodemap=NULL;
 
-	/* check that there are valid nodes available */
-	if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap) != 0) {
+
+	/* check if the node is already enabled */
+	if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
 		DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
 		exit(10);
 	}
-	for (i=0; i<nodemap->num;i++) {
-		if ((nodemap->nodes[i].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) == 0) {
-			break;
-		}
-	}
-	if (i==nodemap->num) {
-		DEBUG(DEBUG_ERR,("No recmaster available, no need to wait for cluster convergence\n"));
+	if (!(nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED)) {
+		DEBUG(DEBUG_ERR,("Node %d is already enabled.\n", options.pnn));
 		return 0;
 	}
 
+	do {
+		ret = ctdb_ctrl_modflags(ctdb, TIMELIMIT(), options.pnn, 0, NODE_FLAGS_PERMANENTLY_DISABLED);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, ("Unable to enable node %u\n", options.pnn));
+			return ret;
+		}
+
+		sleep(1);
 
-	ret = ctdb_ctrl_getrecmaster(ctdb, ctdb, TIMELIMIT(), options.pnn, &recmaster);
+		/* read the nodemap and verify the change took effect */
+		if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
+			DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
+			exit(10);
+		}
+
+	} while (nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED);
+
+	ret = control_ipreallocate(ctdb, argc, argv);
 	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to get recmaster from node %u\n", options.pnn));
-		return ret;
-	}
-
-	/* verify the node exists */
-	if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), recmaster, ctdb, &nodemap) != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
-		exit(10);
-	}
-
-
-	/* check tha there are nodes available that can act as a recmaster */
-	for (i=0; i<nodemap->num; i++) {
-		if (nodemap->nodes[i].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
-			continue;
-		}
-		break;
-	}
-	if (i == nodemap->num) {
-		DEBUG(DEBUG_ERR,("No possible nodes to host addresses.\n"));
-		return 0;
-	}
-
-	/* verify the recovery master is not STOPPED, nor BANNED */
-	if (nodemap->nodes[recmaster].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
-		DEBUG(DEBUG_ERR,("No suitable recmaster found. Try again\n"));
-		retries++;
-		sleep(1);
-		goto again;
-	} 
-
-	
-	/* verify the recovery master is not STOPPED, nor BANNED */
-	if (nodemap->nodes[recmaster].flags & (NODE_FLAGS_DELETED|NODE_FLAGS_BANNED|NODE_FLAGS_STOPPED)) {
-		DEBUG(DEBUG_ERR,("No suitable recmaster found. Try again\n"));
-		retries++;
-		sleep(1);
-		goto again;
-	} 
-
-	ret = ctdb_send_message(ctdb, recmaster, CTDB_SRVID_TAKEOVER_RUN, data);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,("Failed to send ip takeover run request message to %u\n", options.pnn));
-		return -1;
-	}
-
-	tv = timeval_current();
-	/* this loop will terminate when we have received the reply */
-	while (timeval_elapsed(&tv) < 3.0) {	
-		event_loop_once(ctdb->ev);
-	}
-
-	DEBUG(DEBUG_INFO,("Timed out waiting for recmaster ipreallocate. Trying again\n"));
-	retries++;
-	sleep(1);
-	goto again;
-
-	return 0;
-}
-
-
-/*
-  disable a remote node
- */
-static int control_disable(struct ctdb_context *ctdb, int argc, const char **argv)
-{
-	int ret;
-	struct ctdb_node_map *nodemap=NULL;
-
-	do {
-		ret = ctdb_ctrl_modflags(ctdb, TIMELIMIT(), options.pnn, NODE_FLAGS_PERMANENTLY_DISABLED, 0);
-		if (ret != 0) {
-			DEBUG(DEBUG_ERR, ("Unable to disable node %u\n", options.pnn));
-			return ret;
-		}
-
-		sleep(1);
-
-		/* read the nodemap and verify the change took effect */
-		if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
-			DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
-			exit(10);
-		}
-
-	} while (!(nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED));
-	ret = control_ipreallocate(ctdb, argc, argv);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
-		return ret;
-	}
-
-	return 0;
-}
-
-/*
-  enable a disabled remote node
- */
-static int control_enable(struct ctdb_context *ctdb, int argc, const char **argv)
-{
-	int ret;
-
-	struct ctdb_node_map *nodemap=NULL;
-
-	do {
-		ret = ctdb_ctrl_modflags(ctdb, TIMELIMIT(), options.pnn, 0, NODE_FLAGS_PERMANENTLY_DISABLED);
-		if (ret != 0) {
-			DEBUG(DEBUG_ERR, ("Unable to enable node %u\n", options.pnn));
-			return ret;
-		}
-
-		sleep(1);
-
-		/* read the nodemap and verify the change took effect */
-		if (ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &nodemap) != 0) {
-			DEBUG(DEBUG_ERR, ("Unable to get nodemap from local node\n"));
-			exit(10);
-		}
-
-	} while (nodemap->nodes[options.pnn].flags & NODE_FLAGS_PERMANENTLY_DISABLED);
-	ret = control_ipreallocate(ctdb, argc, argv);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
+		DEBUG(DEBUG_ERR, ("IP Reallocate failed on node %u\n", options.pnn));
 		return ret;
 	}
 
@@ -2551,149 +2827,475 @@ static int control_lvsmaster(struct ctdb_context *ctdb, int argc, const char **a
 	int i, ret;
 	int healthy_count = 0;
 
-	ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
-		return ret;
+	ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
+		return ret;
+	}
+
+	capabilities = talloc_array(ctdb, uint32_t, nodemap->num);
+	CTDB_NO_MEMORY(ctdb, capabilities);
+	
+	/* collect capabilities for all connected nodes */
+	for (i=0; i<nodemap->num; i++) {
+		if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
+			continue;
+		}
+		if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
+			continue;
+		}
+	
+		ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), i, &capabilities[i]);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", i));
+			return ret;
+		}
+
+		if (!(capabilities[i] & CTDB_CAP_LVS)) {
+			continue;
+		}
+
+		if (!(nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY)) {
+			healthy_count++;
+		}
+	}
+
+	/* find and show the lvsmaster */
+	for (i=0; i<nodemap->num; i++) {
+		if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
+			continue;
+		}
+		if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
+			continue;
+		}
+		if (!(capabilities[i] & CTDB_CAP_LVS)) {
+			continue;
+		}
+
+		if (healthy_count != 0) {
+			if (nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY) {
+				continue;
+			}
+		}
+
+		if (options.machinereadable){
+			printf("%d\n", i);
+		} else {
+			printf("Node %d is LVS master\n", i);
+		}
+		return 0;
+	}
+
+	printf("There is no LVS master\n");
+	return -1;
+}
+
+/*
+  disable monitoring on a  node
+ */
+static int control_disable_monmode(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	
+	int ret;
+
+	ret = ctdb_ctrl_disable_monmode(ctdb, TIMELIMIT(), options.pnn);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to disable monmode on node %u\n", options.pnn));
+		return ret;
+	}
+	printf("Monitoring mode:%s\n","DISABLED");
+
+	return 0;
+}
+
+/*
+  enable monitoring on a  node
+ */
+static int control_enable_monmode(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	
+	int ret;
+
+	ret = ctdb_ctrl_enable_monmode(ctdb, TIMELIMIT(), options.pnn);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Unable to enable monmode on node %u\n", options.pnn));
+		return ret;
+	}
+	printf("Monitoring mode:%s\n","ACTIVE");
+
+	return 0;
+}
+
+/*
+  display remote list of keys/data for a db
+ */
+static int control_catdb(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	const char *db_name;
+	struct ctdb_db_context *ctdb_db;
+	int ret;
+
+	if (argc < 1) {
+		usage();
+	}
+
+	db_name = argv[0];
+
+
+	if (db_exists(ctdb, db_name)) {
+		DEBUG(DEBUG_ERR,("Database '%s' does not exist\n", db_name));
+		return -1;
+	}
+
+	ctdb_db = ctdb_attach(ctdb, db_name, false, 0);
+
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
+		return -1;
+	}
+
+	/* traverse and dump the cluster tdb */
+	ret = ctdb_dump_db(ctdb_db, stdout);
+	if (ret == -1) {
+		DEBUG(DEBUG_ERR, ("Unable to dump database\n"));
+		DEBUG(DEBUG_ERR, ("Maybe try 'ctdb getdbstatus %s'"
+				  " and 'ctdb getvar AllowUnhealthyDBRead'\n",
+				  db_name));
+		return -1;
+	}
+	talloc_free(ctdb_db);
+
+	printf("Dumped %d records\n", ret);
+	return 0;
+}
+
+/*
+  display the content of a database key
+ */
+static int control_readkey(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	const char *db_name;
+	struct ctdb_db_context *ctdb_db;
+	struct ctdb_record_handle *h;
+	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
+	TDB_DATA key, data;
+
+	if (argc < 2) {
+		usage();
+	}
+
+	db_name = argv[0];
+
+
+	if (db_exists(ctdb, db_name)) {
+		DEBUG(DEBUG_ERR,("Database '%s' does not exist\n", db_name));
+		return -1;
+	}
+
+	ctdb_db = ctdb_attach(ctdb, db_name, false, 0);
+
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
+		return -1;
+	}
+
+	key.dptr  = discard_const(argv[1]);
+	key.dsize = strlen((char *)key.dptr);
+ 
+	h = ctdb_fetch_lock(ctdb_db, tmp_ctx, key, &data);
+	if (h == NULL) {
+		printf("Failed to fetch record '%s' on node %d\n", 
+	       		(const char *)key.dptr, ctdb_get_pnn(ctdb));
+		talloc_free(tmp_ctx);
+		exit(10);
+	}
+
+	printf("Data: size:%d ptr:[%s]\n", (int)data.dsize, data.dptr);
+
+	talloc_free(ctdb_db);
+	talloc_free(tmp_ctx);
+	return 0;
+}
+
+/*
+  display the content of a database key
+ */
+static int control_writekey(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	const char *db_name;
+	struct ctdb_db_context *ctdb_db;
+	struct ctdb_record_handle *h;
+	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
+	TDB_DATA key, data;
+
+	if (argc < 3) {
+		usage();
+	}
+
+	db_name = argv[0];
+
+
+	if (db_exists(ctdb, db_name)) {
+		DEBUG(DEBUG_ERR,("Database '%s' does not exist\n", db_name));
+		return -1;
+	}
+
+	ctdb_db = ctdb_attach(ctdb, db_name, false, 0);
+
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
+		return -1;
+	}
+
+	key.dptr  = discard_const(argv[1]);
+	key.dsize = strlen((char *)key.dptr);
+ 
+	h = ctdb_fetch_lock(ctdb_db, tmp_ctx, key, &data);
+	if (h == NULL) {
+		printf("Failed to fetch record '%s' on node %d\n", 
+	       		(const char *)key.dptr, ctdb_get_pnn(ctdb));
+		talloc_free(tmp_ctx);
+		exit(10);
+	}
+
+	data.dptr  = discard_const(argv[2]);
+	data.dsize = strlen((char *)data.dptr);
+ 
+	if (ctdb_record_store(h, data) != 0) {
+		printf("Failed to store record\n");
+	}
+
+	talloc_free(h);
+	talloc_free(ctdb_db);
+	talloc_free(tmp_ctx);
+	return 0;
+}
+
+/*
+  fetch a record from a persistent database
+ */
+static int control_pfetch(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	const char *db_name;
+	struct ctdb_db_context *ctdb_db;
+	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
+	struct ctdb_transaction_handle *h;
+	TDB_DATA key, data;
+	int fd, ret;
+
+	if (argc < 2) {
+		talloc_free(tmp_ctx);
+		usage();
+	}
+
+	db_name = argv[0];
+
+
+	if (db_exists(ctdb, db_name)) {
+		DEBUG(DEBUG_ERR,("Database '%s' does not exist\n", db_name));
+		talloc_free(tmp_ctx);
+		return -1;
 	}
 
-	capabilities = talloc_array(ctdb, uint32_t, nodemap->num);
-	CTDB_NO_MEMORY(ctdb, capabilities);
-	
-	/* collect capabilities for all connected nodes */
-	for (i=0; i<nodemap->num; i++) {
-		if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
-			continue;
-		}
-		if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
-			continue;
-		}
-	
-		ret = ctdb_ctrl_getcapabilities(ctdb, TIMELIMIT(), i, &capabilities[i]);
-		if (ret != 0) {
-			DEBUG(DEBUG_ERR, ("Unable to get capabilities from node %u\n", i));
-			return ret;
-		}
+	ctdb_db = ctdb_attach(ctdb, db_name, true, 0);
 
-		if (!(capabilities[i] & CTDB_CAP_LVS)) {
-			continue;
-		}
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
 
-		if (!(nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY)) {
-			healthy_count++;
-		}
+	h = ctdb_transaction_start(ctdb_db, tmp_ctx);
+	if (h == NULL) {
+		DEBUG(DEBUG_ERR,("Failed to start transaction on database %s\n", db_name));
+		talloc_free(tmp_ctx);
+		return -1;
 	}
 
-	/* find and show the lvsmaster */
-	for (i=0; i<nodemap->num; i++) {
-		if (nodemap->nodes[i].flags & NODE_FLAGS_INACTIVE) {
-			continue;
-		}
-		if (nodemap->nodes[i].flags & NODE_FLAGS_PERMANENTLY_DISABLED) {
-			continue;
-		}
-		if (!(capabilities[i] & CTDB_CAP_LVS)) {
-			continue;
-		}
+	key.dptr  = discard_const(argv[1]);
+	key.dsize = strlen(argv[1]);
+	ret = ctdb_transaction_fetch(h, tmp_ctx, key, &data);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to fetch record\n"));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
 
-		if (healthy_count != 0) {
-			if (nodemap->nodes[i].flags & NODE_FLAGS_UNHEALTHY) {
-				continue;
-			}
-		}
+	if (data.dsize == 0 || data.dptr == NULL) {
+		DEBUG(DEBUG_ERR,("Record is empty\n"));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
 
-		if (options.machinereadable){
-			printf("%d\n", i);
-		} else {
-			printf("Node %d is LVS master\n", i);
+	if (argc == 3) {
+	  fd = open(argv[2], O_WRONLY|O_CREAT|O_TRUNC, 0600);
+		if (fd == -1) {
+			DEBUG(DEBUG_ERR,("Failed to open output file %s\n", argv[2]));
+			talloc_free(tmp_ctx);
+			return -1;
 		}
-		return 0;
+		write(fd, data.dptr, data.dsize);
+		close(fd);
+	} else {
+		write(1, data.dptr, data.dsize);
 	}
 
-	printf("There is no LVS master\n");
-	return -1;
+	/* abort the transaction */
+	talloc_free(h);
+
+
+	talloc_free(tmp_ctx);
+	return 0;
 }
 
 /*
-  disable monitoring on a  node
+  fetch a record from a tdb-file
  */
-static int control_disable_monmode(struct ctdb_context *ctdb, int argc, const char **argv)
+static int control_tfetch(struct ctdb_context *ctdb, int argc, const char **argv)
 {
-	
-	int ret;
+	const char *tdb_file;
+	TDB_CONTEXT *tdb;
+	TDB_DATA key, data;
+	int fd;
 
-	ret = ctdb_ctrl_disable_monmode(ctdb, TIMELIMIT(), options.pnn);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to disable monmode on node %u\n", options.pnn));
-		return ret;
+	if (argc < 2) {
+		usage();
 	}
-	printf("Monitoring mode:%s\n","DISABLED");
 
-	return 0;
-}
+	tdb_file = argv[0];
 
-/*
-  enable monitoring on a  node
- */
-static int control_enable_monmode(struct ctdb_context *ctdb, int argc, const char **argv)
-{
-	
-	int ret;
+	tdb = tdb_open(tdb_file, 0, 0, O_RDONLY, 0);
+	if (tdb == NULL) {
+		DEBUG(DEBUG_ERR,("Failed to open TDB file %s\n", tdb_file));
+		return -1;
+	}
 
-	ret = ctdb_ctrl_enable_monmode(ctdb, TIMELIMIT(), options.pnn);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to enable monmode on node %u\n", options.pnn));
-		return ret;
+	key.dptr  = discard_const(argv[1]);
+	key.dsize = strlen(argv[1]);
+	data = tdb_fetch(tdb, key);
+	if (data.dptr == NULL || data.dsize < sizeof(struct ctdb_ltdb_header)) {
+		DEBUG(DEBUG_ERR,("Failed to read record %s from tdb %s\n", argv[1], tdb_file));
+		tdb_close(tdb);
+		return -1;
+	}
+
+	tdb_close(tdb);
+
+	if (argc == 3) {
+	  fd = open(argv[2], O_WRONLY|O_CREAT|O_TRUNC, 0600);
+		if (fd == -1) {
+			DEBUG(DEBUG_ERR,("Failed to open output file %s\n", argv[2]));
+			return -1;
+		}
+		write(fd, data.dptr+sizeof(struct ctdb_ltdb_header), data.dsize-sizeof(struct ctdb_ltdb_header));
+		close(fd);
+	} else {
+		write(1, data.dptr+sizeof(struct ctdb_ltdb_header), data.dsize-sizeof(struct ctdb_ltdb_header));
 	}
-	printf("Monitoring mode:%s\n","ACTIVE");
 
 	return 0;
 }
 
 /*
-  display remote list of keys/data for a db
+  write a record to a persistent database
  */
-static int control_catdb(struct ctdb_context *ctdb, int argc, const char **argv)
+static int control_pstore(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	const char *db_name;
 	struct ctdb_db_context *ctdb_db;
-	int ret;
+	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
+	struct ctdb_transaction_handle *h;
+	struct stat st;
+	TDB_DATA key, data;
+	int fd, ret;
 
-	if (argc < 1) {
+	if (argc < 3) {
+		talloc_free(tmp_ctx);
 		usage();
 	}
 
-	db_name = argv[0];
-
+	fd = open(argv[2], O_RDONLY);
+	if (fd == -1) {
+		DEBUG(DEBUG_ERR,("Failed to open file containing record data : %s  %s\n", argv[2], strerror(errno)));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
+	
+	ret = fstat(fd, &st);
+	if (ret == -1) {
+		DEBUG(DEBUG_ERR,("fstat of file %s failed: %s\n", argv[2], strerror(errno)));
+		close(fd);
+		talloc_free(tmp_ctx);
+		return -1;
+	}
 
-	if (db_exists(ctdb, db_name)) {
-		DEBUG(DEBUG_ERR,("Database '%s' does not exist\n", db_name));
+	if (!S_ISREG(st.st_mode)) {
+		DEBUG(DEBUG_ERR,("Not a regular file %s\n", argv[2]));
+		close(fd);
+		talloc_free(tmp_ctx);
 		return -1;
 	}
 
-	ctdb_db = ctdb_attach(ctdb, db_name, false, 0);
+	data.dsize = st.st_size;
+	if (data.dsize == 0) {
+		data.dptr  = NULL;
+	} else {
+		data.dptr = talloc_size(tmp_ctx, data.dsize);
+		if (data.dptr == NULL) {
+			DEBUG(DEBUG_ERR,("Failed to talloc %d of memory to store record data\n", (int)data.dsize));
+			close(fd);
+			talloc_free(tmp_ctx);
+			return -1;
+		}
+		ret = read(fd, data.dptr, data.dsize);
+		if (ret != data.dsize) {
+			DEBUG(DEBUG_ERR,("Failed to read %d bytes of record data\n", (int)data.dsize));
+			close(fd);
+			talloc_free(tmp_ctx);
+			return -1;
+		}
+	}
+	close(fd);
+
+
+	db_name = argv[0];
+
+	ctdb_db = ctdb_attach(ctdb, db_name, true, 0);
 
 	if (ctdb_db == NULL) {
 		DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", db_name));
+		talloc_free(tmp_ctx);
 		return -1;
 	}
 
-	/* traverse and dump the cluster tdb */
-	ret = ctdb_dump_db(ctdb_db, stdout);
-	if (ret == -1) {
-		DEBUG(DEBUG_ERR, ("Unable to dump database\n"));
-		DEBUG(DEBUG_ERR, ("Maybe try 'ctdb getdbstatus %s'"
-				  " and 'ctdb getvar AllowUnhealthyDBRead'\n",
-				  db_name));
+	h = ctdb_transaction_start(ctdb_db, tmp_ctx);
+	if (h == NULL) {
+		DEBUG(DEBUG_ERR,("Failed to start transaction on database %s\n", db_name));
+		talloc_free(tmp_ctx);
 		return -1;
 	}
-	talloc_free(ctdb_db);
 
-	printf("Dumped %d records\n", ret);
+	key.dptr  = discard_const(argv[1]);
+	key.dsize = strlen(argv[1]);
+	ret = ctdb_transaction_store(h, key, data);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to store record\n"));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
+
+	ret = ctdb_transaction_commit(h);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR,("Failed to commit transaction\n"));
+		talloc_free(tmp_ctx);
+		return -1;
+	}
+
+
+	talloc_free(tmp_ctx);
 	return 0;
 }
 
-
 static void log_handler(struct ctdb_context *ctdb, uint64_t srvid, 
 			     TDB_DATA data, void *private_data)
 {
@@ -2738,7 +3340,7 @@ static int control_getlog(struct ctdb_context *ctdb, int argc, const char **argv
 
 	DEBUG(DEBUG_ERR, ("Pulling logs from node %u\n", options.pnn));
 
-	ctdb_set_message_handler(ctdb, log_addr.srvid, log_handler, NULL);
+	ctdb_client_set_message_handler(ctdb, log_addr.srvid, log_handler, NULL);
 	sleep(1);
 
 	DEBUG(DEBUG_ERR,("Listen for response on %d\n", (int)log_addr.srvid));
@@ -3218,28 +3820,6 @@ static int control_setdebug(struct ctdb_context *ctdb, int argc, const char **ar
 }
 
 
-/*
-  freeze a node
- */
-static int control_freeze(struct ctdb_context *ctdb, int argc, const char **argv)
-{
-	int ret;
-	uint32_t priority;
-	
-	if (argc == 1) {
-		priority = strtol(argv[0], NULL, 0);
-	} else {
-		priority = 0;
-	}
-	DEBUG(DEBUG_ERR,("Freeze by priority %u\n", priority));
-
-	ret = ctdb_ctrl_freeze_priority(ctdb, TIMELIMIT(), options.pnn, priority);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("Unable to freeze node %u\n", options.pnn));
-	}		
-	return 0;
-}
-
 /*
   thaw a node
  */
@@ -3585,8 +4165,9 @@ static int control_restoredb(struct ctdb_context *ctdb, int argc, const char **a
 	uint32_t generation;
 	struct tm *tm;
 	char tbuf[100];
+	char *dbname;
 
-	if (argc != 1) {
+	if (argc < 1 || argc > 2) {
 		DEBUG(DEBUG_ERR,("Invalid arguments\n"));
 		return -1;
 	}
@@ -3605,6 +4186,11 @@ static int control_restoredb(struct ctdb_context *ctdb, int argc, const char **a
 		return -1;
 	}
 
+	dbname = discard_const(dbhdr.name);
+	if (argc == 2) {
+		dbname = discard_const(argv[1]);
+	}
+
 	outdata.dsize = dbhdr.size;
 	outdata.dptr = talloc_size(tmp_ctx, outdata.dsize);
 	if (outdata.dptr == NULL) {
@@ -3619,12 +4205,12 @@ static int control_restoredb(struct ctdb_context *ctdb, int argc, const char **a
 	tm = localtime(&dbhdr.timestamp);
 	strftime(tbuf,sizeof(tbuf)-1,"%Y/%m/%d %H:%M:%S", tm);
 	printf("Restoring database '%s' from backup @ %s\n",
-		dbhdr.name, tbuf);
+		dbname, tbuf);
 
 
-	ctdb_db = ctdb_attach(ctdb, dbhdr.name, dbhdr.persistent, 0);
+	ctdb_db = ctdb_attach(ctdb, dbname, dbhdr.persistent, 0);
 	if (ctdb_db == NULL) {
-		DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", dbhdr.name));
+		DEBUG(DEBUG_ERR,("Unable to attach to database '%s'\n", dbname));
 		talloc_free(tmp_ctx);
 		return -1;
 	}
@@ -4003,50 +4589,6 @@ static int control_wipedb(struct ctdb_context *ctdb, int argc,
 	return 0;
 }
 
-/*
- * set flags of a node in the nodemap
- */
-static int control_setflags(struct ctdb_context *ctdb, int argc, const char **argv)
-{
-	int ret;
-	int32_t status;
-	int node;
-	int flags;
-	TDB_DATA data;
-	struct ctdb_node_flag_change c;
-
-	if (argc != 2) {
-		usage();
-		return -1;
-	}
-
-	if (sscanf(argv[0], "%d", &node) != 1) {
-		DEBUG(DEBUG_ERR, ("Badly formed node\n"));
-		usage();
-		return -1;
-	}
-	if (sscanf(argv[1], "0x%x", &flags) != 1) {
-		DEBUG(DEBUG_ERR, ("Badly formed flags\n"));
-		usage();
-		return -1;
-	}
-
-	c.pnn       = node;
-	c.old_flags = 0;
-	c.new_flags = flags;
-
-	data.dsize = sizeof(c);
-	data.dptr = (unsigned char *)&c;
-
-	ret = ctdb_control(ctdb, options.pnn, 0, CTDB_CONTROL_MODIFY_FLAGS, 0, 
-			   data, NULL, NULL, &status, NULL, NULL);
-	if (ret != 0 || status != 0) {
-		DEBUG(DEBUG_ERR,("Failed to modify flags\n"));
-		return -1;
-	}
-	return 0;
-}
-
 /*
   dump memory usage
  */
@@ -4098,13 +4640,13 @@ static int control_rddumpmemory(struct ctdb_context *ctdb, int argc, const char
 	/* register a message port for receiveing the reply so that we
 	   can receive the reply
 	*/
-	ctdb_set_message_handler(ctdb, rd.srvid, mem_dump_handler, NULL);
+	ctdb_client_set_message_handler(ctdb, rd.srvid, mem_dump_handler, NULL);
 
 
 	data.dptr = (uint8_t *)&rd;
 	data.dsize = sizeof(rd);
 
-	ret = ctdb_send_message(ctdb, options.pnn, CTDB_SRVID_MEM_DUMP, data);
+	ret = ctdb_client_send_message(ctdb, options.pnn, CTDB_SRVID_MEM_DUMP, data);
 	if (ret != 0) {
 		DEBUG(DEBUG_ERR,("Failed to send memdump request message to %u\n", options.pnn));
 		return -1;
@@ -4136,7 +4678,7 @@ static int control_msgsend(struct ctdb_context *ctdb, int argc, const char **arg
 	data.dptr = (uint8_t *)discard_const(argv[1]);
 	data.dsize= strlen(argv[1]);
 
-	ret = ctdb_send_message(ctdb, CTDB_BROADCAST_CONNECTED, srvid, data);
+	ret = ctdb_client_send_message(ctdb, CTDB_BROADCAST_CONNECTED, srvid, data);
 	if (ret != 0) {
 		DEBUG(DEBUG_ERR,("Failed to send memdump request message to %u\n", options.pnn));
 		return -1;
@@ -4171,7 +4713,7 @@ static int control_msglisten(struct ctdb_context *ctdb, int argc, const char **a
 
 	/* register a message port and listen for messages
 	*/
-	ctdb_set_message_handler(ctdb, srvid, msglisten_handler, NULL);
+	ctdb_client_set_message_handler(ctdb, srvid, msglisten_handler, NULL);
 	printf("Listening for messages on srvid:%d\n", (int)srvid);
 
 	while (1) {	
@@ -4183,60 +4725,35 @@ static int control_msglisten(struct ctdb_context *ctdb, int argc, const char **a
 
 /*
   list all nodes in the cluster
-  if the daemon is running, we read the data from the daemon.
-  if the daemon is not running we parse the nodes file directly
+  we parse the nodes file directly
  */
 static int control_listnodes(struct ctdb_context *ctdb, int argc, const char **argv)
 {
-	int i, ret;
-	struct ctdb_node_map *nodemap=NULL;
+	TALLOC_CTX *mem_ctx = talloc_new(NULL);
+	struct pnn_node *pnn_nodes;
+	struct pnn_node *pnn_node;
 
-	if (ctdb != NULL) {
-		ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(), options.pnn, ctdb, &nodemap);
-		if (ret != 0) {
-			DEBUG(DEBUG_ERR, ("Unable to get nodemap from node %u\n", options.pnn));
-			return ret;
-		}
+	pnn_nodes = read_nodes_file(mem_ctx);
+	if (pnn_nodes == NULL) {
+		DEBUG(DEBUG_ERR,("Failed to read nodes file\n"));
+		talloc_free(mem_ctx);
+		return -1;
+	}
 
-		for(i=0;i<nodemap->num;i++){
-			if (nodemap->nodes[i].flags & NODE_FLAGS_DELETED) {
-				continue;
-			}
-			if (options.machinereadable){
-				printf(":%d:%s:\n", nodemap->nodes[i].pnn, ctdb_addr_to_str(&nodemap->nodes[i].addr));
-			} else {
-				printf("%s\n", ctdb_addr_to_str(&nodemap->nodes[i].addr));
-			}
-		}
-	} else {
-		TALLOC_CTX *mem_ctx = talloc_new(NULL);
-		struct pnn_node *pnn_nodes;
-		struct pnn_node *pnn_node;
-	
-		pnn_nodes = read_nodes_file(mem_ctx);
-		if (pnn_nodes == NULL) {
-			DEBUG(DEBUG_ERR,("Failed to read nodes file\n"));
+	for(pnn_node=pnn_nodes;pnn_node;pnn_node=pnn_node->next) {
+		ctdb_sock_addr addr;
+		if (parse_ip(pnn_node->addr, NULL, 63999, &addr) == 0) {
+			DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s' in nodes file\n", pnn_node->addr));
 			talloc_free(mem_ctx);
 			return -1;
 		}
-
-		for(pnn_node=pnn_nodes;pnn_node;pnn_node=pnn_node->next) {
-			ctdb_sock_addr addr;
-
-			if (parse_ip(pnn_node->addr, NULL, 63999, &addr) == 0) {
-				DEBUG(DEBUG_ERR,("Wrongly formed ip address '%s' in nodes file\n", pnn_node->addr));
-				talloc_free(mem_ctx);
-				return -1;
-			}
-
-			if (options.machinereadable){
-				printf(":%d:%s:\n", pnn_node->pnn, pnn_node->addr);
-			} else {
-				printf("%s\n", pnn_node->addr);
-			}
+		if (options.machinereadable){
+			printf(":%d:%s:\n", pnn_node->pnn, pnn_node->addr);
+		} else {
+			printf("%s\n", pnn_node->addr);
 		}
-		talloc_free(mem_ctx);
 	}
+	talloc_free(mem_ctx);
 
 	return 0;
 }
@@ -4308,6 +4825,7 @@ static const struct {
 	{ "listvars",        control_listvars,          true,	false,  "list tunable variables"},
 	{ "statistics",      control_statistics,        false,	false, "show statistics" },
 	{ "statisticsreset", control_statistics_reset,  true,	false,  "reset statistics"},
+	{ "stats",           control_stats,             false,	false,  "show rolling statistics", "[number of history records]" },
 	{ "ip",              control_ip,                false,	false,  "show which public ip's that ctdb manages" },
 	{ "ipinfo",          control_ipinfo,            true,	false,  "show details about a public ip that ctdb manages", "<ip>" },
 	{ "ifaces",          control_ifaces,            true,	false,  "show which interfaces that ctdb manages" },
@@ -4340,14 +4858,17 @@ static const struct {
 	{ "showban",         control_showban,           true,	false,  "show ban information"},
 	{ "shutdown",        control_shutdown,          true,	false,  "shutdown ctdbd" },
 	{ "recover",         control_recover,           true,	false,  "force recovery" },
+	{ "sync", 	     control_ipreallocate,      true,	false,  "wait until ctdbd has synced all state changes" },
 	{ "ipreallocate",    control_ipreallocate,      true,	false,  "force the recovery daemon to perform a ip reallocation procedure" },
-	{ "freeze",          control_freeze,            true,	false,  "freeze databases", "[priority:1-3]" },
 	{ "thaw",            control_thaw,              true,	false,  "thaw databases", "[priority:1-3]" },
 	{ "isnotrecmaster",  control_isnotrecmaster,    false,	false,  "check if the local node is recmaster or not" },
 	{ "killtcp",         kill_tcp,                  false,	false, "kill a tcp connection.", "<srcip:port> <dstip:port>" },
 	{ "gratiousarp",     control_gratious_arp,      false,	false, "send a gratious arp", "<ip> <interface>" },
 	{ "tickle",          tickle_tcp,                false,	false, "send a tcp tickle ack", "<srcip:port> <dstip:port>" },
-	{ "gettickles",      control_get_tickles,       false,	false, "get the list of tickles registered for this ip", "<ip>" },
+	{ "gettickles",      control_get_tickles,       false,	false, "get the list of tickles registered for this ip", "<ip> [<port>]" },
+	{ "addtickle",       control_add_tickle,        false,	false, "add a tickle for this ip", "<ip>:<port> <ip>:<port>" },
+
+	{ "deltickle",       control_del_tickle,        false,	false, "delete a tickle from this ip", "<ip>:<port> <ip>:<port>" },
 
 	{ "regsrvid",        regsrvid,			false,	false, "register a server id", "<pnn> <type> <id>" },
 	{ "unregsrvid",      unregsrvid,		false,	false, "unregister a server id", "<pnn> <type> <id>" },
@@ -4362,11 +4883,10 @@ static const struct {
 	{ "delip",           control_delip,		false,	false, "delete an ip address from a node", "<ip>"},
 	{ "eventscript",     control_eventscript,	true,	false, "run the eventscript with the given parameters on a node", "<arguments>"},
 	{ "backupdb",        control_backupdb,          false,	false, "backup the database into a file.", "<database> <file>"},
-	{ "restoredb",        control_restoredb,        false,	false, "restore the database from a file.", "<file>"},
+	{ "restoredb",        control_restoredb,        false,	false, "restore the database from a file.", "<file> [dbname]"},
 	{ "dumpdbbackup",    control_dumpdbbackup,      false,	true,  "dump database backup from a file.", "<file>"},
 	{ "wipedb",           control_wipedb,        false,	false, "wipe the contents of a database.", "<dbname>"},
 	{ "recmaster",        control_recmaster,        false,	false, "show the pnn for the recovery master."},
-	{ "setflags",        control_setflags,          false,	false, "set flags for a node in the nodemap.", "<node> <flags>"},
 	{ "scriptstatus",    control_scriptstatus,  false,	false, "show the status of the monitoring scripts (or all scripts)", "[all]"},
 	{ "enablescript",     control_enablescript,  false,	false, "enable an eventscript", "<script>"},
 	{ "disablescript",    control_disablescript,  false,	false, "disable an eventscript", "<script>"},
@@ -4381,6 +4901,12 @@ static const struct {
 	{ "getdbprio",        control_getdbprio,	false,	false, "Get DB priority", "<dbid>"},
 	{ "msglisten",        control_msglisten,	false,	false, "Listen on a srvid port for messages", "<msg srvid>"},
 	{ "msgsend",          control_msgsend,	false,	false, "Send a message to srvid", "<srvid> <message>"},
+	{ "sync", 	     control_ipreallocate,      false,	false,  "wait until ctdbd has synced all state changes" },
+	{ "pfetch", 	     control_pfetch,      	false,	false,  "fetch a record from a persistent database", "<db> <key> [<file>]" },
+	{ "pstore", 	     control_pstore,      	false,	false,  "write a record to a persistent database", "<db> <key> <file containing record>" },
+	{ "tfetch", 	     control_tfetch,      	false,	true,  "fetch a record from a [c]tdb-file", "<tdb-file> <key> [<file>]" },
+	{ "readkey", 	     control_readkey,      	true,	false,  "read the content off a database key", "<tdb-file> <key>" },
+	{ "writekey", 	     control_writekey,      	true,	false,  "write to a database key", "<tdb-file> <key> <value>" },
 };
 
 /*
@@ -4394,6 +4920,7 @@ static void usage(void)
 "Options:\n" \
 "   -n <node>          choose node number, or 'all' (defaults to local node)\n"
 "   -Y                 generate machinereadable output\n"
+"   -v                 generate verbose output\n"
 "   -t <timelimit>     set timelimit for control in seconds (default %u)\n", options.timelimit);
 	printf("Controls:\n");
 	for (i=0;i<ARRAY_SIZE(ctdb_commands);i++) {
@@ -4425,6 +4952,7 @@ int main(int argc, const char *argv[])
 		{ "timelimit", 't', POPT_ARG_INT, &options.timelimit, 0, "timelimit", "integer" },
 		{ "node",      'n', POPT_ARG_STRING, &nodestring, 0, "node", "integer|all" },
 		{ "machinereadable", 'Y', POPT_ARG_NONE, &options.machinereadable, 0, "enable machinereadable output", NULL },
+		{ "verbose",    'v', POPT_ARG_NONE, &options.verbose, 0, "enable verbose output", NULL },
 		{ "maxruntime", 'T', POPT_ARG_INT, &options.maxruntime, 0, "die if runtime exceeds this limit (in seconds)", "integer" },
 		POPT_TABLEEND
 	};
@@ -4491,6 +5019,11 @@ int main(int argc, const char *argv[])
 	control = extra_argv[0];
 
 	ev = event_context_init(NULL);
+	if (!ev) {
+		DEBUG(DEBUG_ERR, ("Failed to initialize event system\n"));
+		exit(1);
+	}
+	tevent_loop_allow_nesting(ev);
 
 	for (i=0;i<ARRAY_SIZE(ctdb_commands);i++) {
 		if (strcmp(control, ctdb_commands[i].name) == 0) {
@@ -4500,15 +5033,26 @@ int main(int argc, const char *argv[])
 				close(2);
 			}
 
-			/* initialise ctdb */
-			ctdb = ctdb_cmdline_client(ev);
-
 			if (ctdb_commands[i].without_daemon == false) {
+				const char *socket_name;
+
+				/* initialise ctdb */
+				ctdb = ctdb_cmdline_client(ev);
+
 				if (ctdb == NULL) {
 					DEBUG(DEBUG_ERR, ("Failed to init ctdb\n"));
 					exit(1);
 				}
 
+				/* initialize a libctdb connection as well */
+				socket_name = ctdb_get_socketname(ctdb);
+				ctdb_connection = ctdb_connect(socket_name,
+						       ctdb_log_file, stderr);
+				if (ctdb_connection == NULL) {
+					fprintf(stderr, "Failed to connect to daemon from libctdb\n");
+					exit(1);
+				}				
+			
 				/* verify the node exists */
 				verify_node(ctdb);