ctdb-daemon: Add control CTDB_CONTROL_DB_DETACH

This detaches specified database from all the nodes.

Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>

4 files changed, 91 insertions, 0 deletions

diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 2206f4edfd9..81759786dd1 100644
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -900,6 +900,8 @@ int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
 			       bool persistent, uint32_t client_id,
 			       struct ctdb_req_control *c,
 			       bool *async_reply);
+int32_t ctdb_control_db_detach(struct ctdb_context *ctdb, TDB_DATA indata,
+			       uint32_t client_id);
 
 int ctdb_daemon_set_call(struct ctdb_context *ctdb, uint32_t db_id,
 			 ctdb_fn_t fn, int id);
diff --git a/ctdb/include/ctdb_protocol.h b/ctdb/include/ctdb_protocol.h
index fa03a4b2f13..f2d68f4c3a3 100644
--- a/ctdb/include/ctdb_protocol.h
+++ b/ctdb/include/ctdb_protocol.h
@@ -404,6 +404,7 @@ enum ctdb_controls {CTDB_CONTROL_PROCESS_EXISTS          = 0,
 		    CTDB_CONTROL_RECEIVE_RECORDS	 = 136,
 		    CTDB_CONTROL_IPREALLOCATED		 = 137,
 		    CTDB_CONTROL_GET_RUNSTATE		 = 138,
+		    CTDB_CONTROL_DB_DETACH		 = 139,
 };
 
 /*
diff --git a/ctdb/server/ctdb_control.c b/ctdb/server/ctdb_control.c
index b56b624f00c..b0220de1505 100644
--- a/ctdb/server/ctdb_control.c
+++ b/ctdb/server/ctdb_control.c
@@ -682,6 +682,9 @@ static int32_t ctdb_control_dispatch(struct ctdb_context *ctdb,
 	case CTDB_CONTROL_RECEIVE_RECORDS:
 		return ctdb_control_receive_records(ctdb, indata, outdata);
 
+	case CTDB_CONTROL_DB_DETACH:
+		return ctdb_control_db_detach(ctdb, indata, client_id);
+
 	default:
 		DEBUG(DEBUG_CRIT,(__location__ " Unknown CTDB control opcode %u\n", opcode));
 		return -1;
diff --git a/ctdb/server/ctdb_ltdb_server.c b/ctdb/server/ctdb_ltdb_server.c
index aafba42c564..3d8772fc887 100644
--- a/ctdb/server/ctdb_ltdb_server.c
+++ b/ctdb/server/ctdb_ltdb_server.c
@@ -1164,6 +1164,91 @@ int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
 	return 0;
 }
 
+/*
+ * a client has asked to detach from a database
+ */
+int32_t ctdb_control_db_detach(struct ctdb_context *ctdb, TDB_DATA indata,
+			       uint32_t client_id)
+{
+	uint32_t db_id;
+	struct ctdb_db_context *ctdb_db;
+	struct ctdb_client *client = NULL;
+
+	db_id = *(uint32_t *)indata.dptr;
+	ctdb_db = find_ctdb_db(ctdb, db_id);
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR, ("Invalid dbid 0x%08x in DB detach\n",
+				  db_id));
+		return -1;
+	}
+
+	if (ctdb->tunable.allow_client_db_attach == 0) {
+		DEBUG(DEBUG_ERR, ("DB detach from database %s denied by "
+				  "tunable AllowClientDBAccess == 0\n",
+				  ctdb_db->db_name));
+		return -1;
+	}
+
+	if (ctdb_db->persistent) {
+		DEBUG(DEBUG_ERR, ("DB detach from persistent database %s "
+				  "denied\n", ctdb_db->db_name));
+		return -1;
+	}
+
+	/* Cannot detach from database when in recovery */
+	if (ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE) {
+		DEBUG(DEBUG_ERR, ("DB detach denied while in recovery\n"));
+		return -1;
+	}
+
+	/* If a control comes from a client, then broadcast it to all nodes.
+	 * Do the actual detach only if the control comes from other daemons.
+	 */
+	if (client_id != 0) {
+		client = ctdb_reqid_find(ctdb, client_id, struct ctdb_client);
+		if (client != NULL) {
+			/* forward the control to all the nodes */
+			ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_ALL, 0,
+						 CTDB_CONTROL_DB_DETACH, 0,
+						 CTDB_CTRL_FLAG_NOREPLY,
+						 indata, NULL, NULL);
+			return 0;
+		}
+		DEBUG(DEBUG_ERR, ("Client has gone away. Failing DB detach "
+				  "for database '%s'\n", ctdb_db->db_name));
+		return -1;
+	}
+
+	/* Disable vacuuming and drop all vacuuming data */
+	talloc_free(ctdb_db->vacuum_handle);
+	talloc_free(ctdb_db->delete_queue);
+
+	/* Terminate any deferred fetch */
+	talloc_free(ctdb_db->deferred_fetch);
+
+	/* Terminate any traverses */
+	while (ctdb_db->traverse) {
+		talloc_free(ctdb_db->traverse);
+	}
+
+	/* Terminate any revokes */
+	while (ctdb_db->revokechild_active) {
+		talloc_free(ctdb_db->revokechild_active);
+	}
+
+	/* Free readonly tracking database */
+	if (ctdb_db->readonly) {
+		talloc_free(ctdb_db->rottdb);
+	}
+
+	DLIST_REMOVE(ctdb->db_list, ctdb_db);
+
+	DEBUG(DEBUG_NOTICE, ("Detached from database '%s'\n",
+			     ctdb_db->db_name));
+	talloc_free(ctdb_db);
+
+	return 0;
+}
 
 /*
   attach to all existing persistent databases