traverse: add a flag to enable transferring empty records in cluster wide traverse

This will be useful for also printing information about empty/deleted
records in "ctdb catdb", e.g. for debugging vacuuming issues.

(This used to be ctdb commit ddc5da3a0df7701934404192a0a0aa659a806acb)

4 files changed, 14 insertions, 1 deletions

diff --git a/ctdb/client/ctdb_client.c b/ctdb/client/ctdb_client.c
index 1d63fa0ad7..307c19ed55 100644
--- a/ctdb/client/ctdb_client.c
+++ b/ctdb/client/ctdb_client.c
@@ -2071,6 +2071,7 @@ int ctdb_traverse(struct ctdb_db_context *ctdb_db, ctdb_traverse_func fn, void *
 	t.db_id = ctdb_db->db_id;
 	t.srvid = srvid;
 	t.reqid = 0;
+	t.withemptyrecords = false;
 
 	data.dptr = (uint8_t *)&t;
 	data.dsize = sizeof(t);
diff --git a/ctdb/include/ctdb_protocol.h b/ctdb/include/ctdb_protocol.h
index f83399c964..f404ec7905 100644
--- a/ctdb/include/ctdb_protocol.h
+++ b/ctdb/include/ctdb_protocol.h
@@ -581,6 +581,7 @@ struct ctdb_traverse_start {
 	uint32_t db_id;
 	uint32_t reqid;
 	uint64_t srvid;
+	bool withemptyrecords;
 };
 
 /*
diff --git a/ctdb/libctdb/ctdb.c b/ctdb/libctdb/ctdb.c
index 2f694a1c34..a525d6fb6d 100644
--- a/ctdb/libctdb/ctdb.c
+++ b/ctdb/libctdb/ctdb.c
@@ -1127,6 +1127,7 @@ static void traverse_msghnd_cb(struct ctdb_connection *ctdb,
 	t.db_id = ctdb_db->id;
 	t.srvid = state->srvid;
 	t.reqid = 0;
+	t.withemptyrecords = false;
 
 	state->handle = new_ctdb_control_request(ctdb,
 				CTDB_CONTROL_TRAVERSE_START,
diff --git a/ctdb/server/ctdb_traverse.c b/ctdb/server/ctdb_traverse.c
index 59ec9b2c75..d719f6f410 100644
--- a/ctdb/server/ctdb_traverse.c
+++ b/ctdb/server/ctdb_traverse.c
@@ -43,6 +43,7 @@ struct ctdb_traverse_local_handle {
 	ctdb_traverse_fn_t callback;
 	struct timeval start_time;
 	struct ctdb_queue *queue;
+	bool withemptyrecords;
 };
 
 /*
@@ -96,7 +97,9 @@ static int ctdb_traverse_local_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DAT
 
 	if (h->ctdb_db->persistent == 0) {
 		/* filter out zero-length records */
-		if (data.dsize <= sizeof(struct ctdb_ltdb_header)) {
+		if (!h->withemptyrecords &&
+		    data.dsize <= sizeof(struct ctdb_ltdb_header))
+		{
 			return 0;
 		}
 
@@ -125,6 +128,7 @@ struct traverse_all_state {
 	uint32_t srcnode;
 	uint32_t client_reqid;
 	uint64_t srvid;
+	bool withemptyrecords;
 };
 
 /*
@@ -167,6 +171,7 @@ static struct ctdb_traverse_local_handle *ctdb_traverse_local(struct ctdb_db_con
 	h->ctdb_db = ctdb_db;
 	h->client_reqid = all_state->client_reqid;
 	h->srvid = all_state->srvid;
+	h->withemptyrecords = all_state->withemptyrecords;
 
 	if (h->child == 0) {
 		/* start the traverse in the child */
@@ -227,6 +232,7 @@ struct ctdb_traverse_all {
 	uint32_t pnn;
 	uint32_t client_reqid;
 	uint64_t srvid;
+	bool withemptyrecords;
 };
 
 /* called when a traverse times out */
@@ -249,6 +255,7 @@ struct traverse_start_state {
 	uint32_t reqid;
 	uint32_t db_id;
 	uint64_t srvid;
+	bool withemptyrecords;
 };
 
 
@@ -290,6 +297,7 @@ static struct ctdb_traverse_all_handle *ctdb_daemon_traverse_all(struct ctdb_db_
 	r.pnn   = ctdb->pnn;
 	r.client_reqid = start_state->reqid;
 	r.srvid = start_state->srvid;
+	r.withemptyrecords = start_state->withemptyrecords;
 
 	data.dptr = (uint8_t *)&r;
 	data.dsize = sizeof(r);
@@ -411,6 +419,7 @@ int32_t ctdb_control_traverse_all(struct ctdb_context *ctdb, TDB_DATA data, TDB_
 	state->ctdb = ctdb;
 	state->client_reqid = c->client_reqid;
 	state->srvid = c->srvid;
+	state->withemptyrecords = c->withemptyrecords;
 
 	state->h = ctdb_traverse_local(ctdb_db, traverse_all_callback, state);
 	if (state->h == NULL) {
@@ -594,6 +603,7 @@ int32_t ctdb_control_traverse_start(struct ctdb_context *ctdb, TDB_DATA data,
 	state->srvid = d->srvid;
 	state->db_id = d->db_id;
 	state->ctdb = ctdb;
+	state->withemptyrecords = d->withemptyrecords;
 
 	state->h = ctdb_daemon_traverse_all(ctdb_db, traverse_start_callback, state);
 	if (state->h == NULL) {