Start/stop metadata replication on PBD plug/unplug.

Signed-off-by: John Else <[email protected]>

Author: johnelse

ocaml/xapi/xapi_pbd.ml

@@ -15,6 +15,7 @@
  * @group XenAPI functions
  *)
 
+open Client
 open Db_filter
 open Db_filter_types
 
@@ -79,12 +80,31 @@ let abort_if_storage_attached_to_protected_vms ~__context ~self =
       ) protected_vms
   end
 
+let find_metadata_vdis ~__context ~sr =
+	let pool = Helpers.get_pool ~__context in
+	List.filter
+		(fun vdi ->
+			Db.VDI.get_type ~__context ~self:vdi = `metadata &&
+			Db.VDI.get_metadata_of_pool ~__context ~self:vdi = pool)
+		(Db.SR.get_VDIs ~__context ~self:sr)
+
 let plug ~__context ~self =
 	let currently_attached = Db.PBD.get_currently_attached ~__context ~self in
 		if not currently_attached then
 			begin
 				let sr = Db.PBD.get_SR ~__context ~self in
 				Storage_access.SR.attach ~__context ~self:sr;
+				(* Try to re-enable metadata replication to all suitable VDIs. *)
+				try
+					List.iter
+						(fun vdi ->
+							debug "Automatically re-enabling database replication to VDI %s" (Ref.string_of vdi);
+							Xapi_vdi_helpers.enable_database_replication ~__context ~vdi)
+						(find_metadata_vdis ~__context ~sr)
+				with Api_errors.Server_error(code, _) when code = Api_errors.no_more_redo_logs_allowed ->
+					(* The redo log limit was reached - don't throw an exception since automatic *)
+					(* re-enabling of database replication is best-effort. *)
+					debug "Metadata is being replicated to the maximum allowed number of VDIs."
 			end
 
 let unplug ~__context ~self =
@@ -106,6 +126,13 @@ let unplug ~__context ~self =
 				then raise (Api_errors.Server_error(Api_errors.ha_is_enabled, []))
 			end;
 
+			(* Disable metadata replication to VDIs in the SR. *)
+			List.iter
+				(fun vdi ->
+					debug "Automatically disabling database replication to VDI %s" (Ref.string_of vdi);
+					Xapi_vdi_helpers.disable_database_replication ~__context ~vdi)
+				(find_metadata_vdis ~__context ~sr);
+
 			let vdis = get_active_vdis_by_pbd ~__context ~self in
 			if List.length vdis > 0 
 			then raise (Api_errors.Server_error(Api_errors.vdi_in_use,List.map Ref.string_of vdis));

ocaml/xapi/xapi_sr.ml

@@ -479,7 +479,14 @@ let disable_database_replication ~__context ~sr =
 		(Db.SR.get_VDIs ~__context ~self:sr)
 	in
 	List.iter
-		(fun vdi -> Xapi_vdi_helpers.disable_database_replication ~__context ~vdi)
+		(fun vdi ->
+			Xapi_vdi_helpers.disable_database_replication ~__context ~vdi;
+			(* The VDI may have VBDs hanging around other than those created by the database replication code. *)
+			(* They must be destroyed before the VDI can be destroyed. *)
+			Xapi_vdi_helpers.destroy_all_vbds ~__context ~vdi;
+			Helpers.call_api_functions ~__context (fun rpc session_id ->
+			 	Client.VDI.destroy ~rpc ~session_id ~self:vdi)
+		)
 		metadata_vdis
 
 let create_new_blob ~__context ~sr ~name ~mime_type =

ocaml/xapi/xapi_vdi_helpers.ml

@@ -47,12 +47,25 @@ let get_master_dom0 ~__context =
 	let vms = Db.Host.get_resident_VMs ~__context ~self:master in
 	List.hd (List.filter (fun vm -> Db.VM.get_is_control_domain ~__context ~self:vm) vms)
 
+(* Unplug and destroy any existing VBDs owned by the VDI. *)
+let destroy_all_vbds ~__context ~vdi =
+	let existing_vbds = Db.VDI.get_VBDs ~__context ~self:vdi in
+	Helpers.call_api_functions ~__context
+		(fun rpc session_id -> List.iter
+			(fun vbd ->
+				try
+					Client.VBD.unplug ~rpc ~session_id ~self:vbd;
+					Client.VBD.destroy ~rpc ~session_id ~self:vbd
+				with Api_errors.Server_error(code, _) when code = Api_errors.device_already_detached ->
+					Client.VBD.destroy ~rpc ~session_id ~self:vbd)
+			existing_vbds)
+
 (* Create and plug a VBD from the VDI, then create a redo log and point it at the block device. *)
 let enable_database_replication ~__context ~vdi =
 	Mutex.execute redo_log_lifecycle_mutex (fun () ->
 		let name_label = Db.VDI.get_name_label ~__context ~self:vdi in
 		let uuid = Db.VDI.get_uuid ~__context ~self:vdi in
-		debug "Attempting to disable metadata replication on VDI [%s:%s]." name_label uuid;
+		debug "Attempting to enable metadata replication on VDI [%s:%s]." name_label uuid;
 		if Hashtbl.mem metadata_replication vdi then
 			debug "Metadata is already being replicated to VDI [%s:%s]." name_label uuid
 		else begin
@@ -62,6 +75,8 @@ let enable_database_replication ~__context ~vdi =
 				raise (Api_errors.Server_error(Api_errors.no_more_redo_logs_allowed, []));
 			let log = Redo_log.create () in
 			let dom0 = get_master_dom0 ~__context in
+			(* We've established that metadata is not being replicated to this VDI, so it should be safe to do this. *)
+			destroy_all_vbds ~__context ~vdi;
 			(* Create and plug vbd *)
 			let vbd = Helpers.call_api_functions ~__context (fun rpc session_id ->
 				let vbd = Client.VBD.create ~rpc ~session_id ~vM:dom0 ~empty:false ~vDI:vdi
@@ -77,6 +92,7 @@ let enable_database_replication ~__context ~vdi =
 			try
 				Redo_log.enable_block log ("/dev/" ^ device);
 				Redo_log.startup log;
+				Redo_log.flush_db_to_redo_log (Db_ref.get_database (Db_backend.make ()));
 				Hashtbl.add metadata_replication vdi (vbd, log);
 				let vbd_uuid = Db.VBD.get_uuid ~__context ~self:vbd in
 				debug "Redo log started on VBD %s" vbd_uuid