<commit_before>(*
 * Copyright (C) 2006-2009 Citrix Systems Inc.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation; version 2.1 only. with the special
 * exception on linking described in file LICENSE.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *)
(** Functions for implementing 'High Availability' (HA). File is divided into 3 sections:
	+ scripts and functions which form part of the HA subsystem interface
	+ internal API calls used for arming and disarming individual hosts
	+ external API calls (Pool.enable_ha, Pool.disable_ha) used for turning on/off HA pool-wide
	* @group High Availability (HA)
 *)

module D = Debug.Debugger(struct let name="xapi_ha" end)
open D

open Listext
open Stringext
open Threadext
open Pervasiveext
open Forkhelpers
open Client
open Threadext
open Db_filter_types

(* Create a redo_log instance to use for HA. *)
let ha_redo_log = Redo_log.create ~name:"HA redo log" ~state_change_callback:None ~read_only:false

(*********************************************************************************************)
(* Interface with the low-level HA subsystem                                                 *)

let ha_set_pool_state = Filename.concat Fhs.xhadir "ha_set_pool_state"
let ha_start_daemon = Filename.concat Fhs.xhadir "ha_start_daemon"
let ha_stop_daemon = Filename.concat Fhs.xhadir "ha_stop_daemon"
let ha_query_liveset = Filename.concat Fhs.xhadir "ha_query_liveset"
let ha_propose_master = Filename.concat Fhs.xhadir "ha_propose_master"
let ha_disarm_fencing = Filename.concat Fhs.xhadir "ha_disarm_fencing"
let ha_set_excluded = Filename.concat Fhs.xhadir "ha_set_excluded"
let fence_path = Filename.concat Fhs.libexecdir "fence"
(* Unused: let ha_clear_excluded = Filename.concat Fhs.xhadir "ha_clear_excluded" *)

(** The xHA scripts throw these exceptions: *)
exception Xha_error of Xha_errno.code

(** Only call one xHA script at a time *)
let ha_script_m = Mutex.create ()

let call_script ?log_successful_output script args =
	try
		Mutex.execute ha_script_m (fun () -> Helpers.call_script ?log_successful_output script args)
	with Forkhelpers.Spawn_internal_error(stderr, stdout, Unix.WEXITED n) ->
		let code = Xha_errno.of_int n in
		warn "%s %s returned %s (%s)" script (String.concat " " args)
			(Xha_errno.to_string code) (Xha_errno.to_description_string code);
		raise (Xha_error code)

(** Returns the current live set info *)
let query_liveset() =
	let txt = call_script ~log_successful_output:false ha_query_liveset [] in
	Xha_interface.LiveSetInformation.of_xml_string txt

(** Returns true if this node has statefile access *)
let i_have_statefile_access () =
	try
		let liveset = query_liveset () in
		let hosts = liveset.Xha_interface.LiveSetInformation.hosts in
		let me = Hashtbl.find hosts liveset.Xha_interface.LiveSetInformation.local_host_id in
		me.Xha_interface.LiveSetInformation.Host.state_file_access && not (me.Xha_interface.LiveSetInformation.Host.state_file_corrupted)
	with e ->
		info "Caught exception querying liveset; assuming we have no statefile access: %s" (ExnHelper.string_of_exn e);
		false

(** Returns true if this node is allowed to be the master *)
let propose_master () =
	try
		let result = call_script ha_propose_master [] in
		String.rtrim result = "TRUE"
	with Xha_error e ->
		error "ha_propose_master threw unexpected exception: %s" (Xha_errno.to_string e);
		false

(** Returns true if local failover decisions have not been disabled on this node *)
let local_failover_decisions_are_ok () =
	try not(bool_of_string (Localdb.get Constants.ha_disable_failover_decisions)) with _ -> true

(** Since the liveset info doesn't include the host IP address, we persist these ourselves *)
let write_uuid_to_ip_mapping ~__context =
	let table = List.map (fun (_, host) -> host.API.host_uuid, host.API.host_address) (Db.Host.get_all_records ~__context) in
	let v = String_marshall_helper.map (fun x -> x) (fun x -> x) table in
	Localdb.put Constants.ha_peers v

(** Since the liveset info doesn't include the host IP address, we persist these ourselves *)
let get_uuid_to_ip_mapping () =
	let v = Localdb.get Constants.ha_peers in
	String_unmarshall_helper.map (fun x -> x) (fun x -> x) v

(** Without using the Pool's database, returns the IP address of a particular host
	named by UUID. *)
let address_of_host_uuid uuid =
	let table = get_uuid_to_ip_mapping () in
	if not(List.mem_assoc uuid table) then begin
		error "Failed to find the IP address of host UUID %s" uuid;
		raise Not_found
	end else List.assoc uuid table

(** Without using the Pool's database, returns the UUID of a particular host named by
	heartbeat IP address. This is only necesary because the liveset info doesn't include
	the host IP address *)
let uuid_of_host_address address =
	let table = List.map (fun (k, v) -> v, k) (get_uuid_to_ip_mapping ()) in
	if not(List.mem_assoc address table) then begin
		error "Failed to find the UUID address of host with address %s" address;
		raise Not_found
	end else List.assoc address table

(** Called in two circumstances:
	1. When I started up I thought I was the master but my proposal was rejected by the
	heartbeat component.
	2. I was happily running as someone's slave but they left the liveset.
 *)
let on_master_failure () =
	(* The plan is: keep asking if I should be the master. If I'm rejected then query the
	   live set and see if someone else has been marked as master, if so become a slave of them. *)

	let become_master () =
		info "This node will become the master";
		Xapi_pool_transition.become_master ();
		info "Waiting for server restart";
		while true do Thread.delay 3600. done in
	let become_slave_of uuid =
		let address = address_of_host_uuid uuid in
		info "This node will become the slave of host %s (%s)" uuid address;
		Xapi_pool_transition.become_another_masters_slave address;
		(* XXX CA-16388: prevent blocking *)
		Thread.delay 15.;
		error "Failed to flush and exit properly; forcibly exiting";
		exit Xapi_globs.restart_return_code in

	let finished = ref false in
	while not !finished do
		(* When HA is disabled without the statefile we set a flag to indicate that this node
		   cannot transition to master automatically on boot. This is to prevent failures during
		   the 'disarm fencing' step which cause some nodes to not fence themselves when they should. *)
		if local_failover_decisions_are_ok () && propose_master () then begin
			info "ha_propose_master succeeded";
			become_master ();
			finished := true
		end else begin
			if local_failover_decisions_are_ok ()
			then info "ha_propose_master failed: looking for another master"
			else info "ha_can_not_be_master_on_next_boot set: I cannot be master; looking for another master";

			let liveset = query_liveset () in
			match Hashtbl.fold
				(fun uuid host acc ->
					if host.Xha_interface.LiveSetInformation.Host.master
						&& host.Xha_interface.LiveSetInformation.Host.liveness (* CP-25481: a dead host may still have the master lock *)
					then uuid :: acc else acc) liveset.Xha_interface.LiveSetInformation.hosts [] with
					| [] ->
						info "no other master exists yet; waiting 5 seconds and retrying";
						Thread.delay 5.
					| uuid :: [] ->
						become_slave_of (Uuid.string_of_uuid uuid)
					| xs ->
						(* should never happen *)
						error "multiple masters reported: [ %s ]; failing"
							(String.concat "; " (List.map Uuid.string_of_uuid xs));
						failwith "multiple masters"
		end
	done

module Timeouts = struct
	type t = {
		heart_beat_interval: int;
		state_file_interval: int;
		heart_beat_timeout: int;
		state_file_timeout: int;
		heart_beat_watchdog_timeout: int;
		state_file_watchdog_timeout: int;
		boot_join_timeout: int;
		enable_join_timeout: int;
		xapi_healthcheck_timeout: int;
		xapi_healthcheck_interval: int;
		xapi_restart_timeout: int;
		xapi_restart_attempts: int;
	}
	let derive (t: int) =
		(* xHA interface section 4.1.4.1.1 Formula of key timeouts *)
		(* t >= 10 *)
		if t < 10 then failwith "constraint violation: timeout >= 10";
		(* All other values are derived from this single parameter *)
		let heart_beat_interval = (t + 10) / 10 in
		let state_file_interval = heart_beat_interval in
		let heart_beat_timeout = t in
		let state_file_timeout = t in
		let heart_beat_watchdog_timeout = t in
		let state_file_watchdog_timeout = t + 15 in
		let boot_join_timeout = t + 60 in
		let enable_join_timeout = boot_join_timeout in

		{ heart_beat_interval = heart_beat_interval;
		state_file_interval = state_file_interval;
		heart_beat_timeout = heart_beat_timeout;
		state_file_timeout = state_file_timeout;
		heart_beat_watchdog_timeout = heart_beat_watchdog_timeout;
		state_file_watchdog_timeout = state_file_watchdog_timeout;
		boot_join_timeout = boot_join_timeout;
		enable_join_timeout = enable_join_timeout;

		xapi_healthcheck_interval = !Xapi_globs.ha_xapi_healthcheck_interval;
		xapi_healthcheck_timeout = !Xapi_globs.ha_xapi_healthcheck_timeout;
		xapi_restart_attempts = !Xapi_globs.ha_xapi_restart_attempts;
		xapi_restart_timeout = !Xapi_globs.ha_xapi_restart_timeout; (* 180s is max start delay and 60s max shutdown delay in the initscript *)
		}

	(** Returns the base timeout value from which the rest are derived *)
	let get_base_t ~__context =
		let pool = Helpers.get_pool ~__context in
		let other_config = Db.Pool.get_other_config ~__context ~self:pool in
		let configuration = Db.Pool.get_ha_configuration ~__context ~self:pool in
		(* xHA built-in default is 30s. We've bumped ours to 40s to make TC7710/CA-17639 happier *)
		(* We then bumped it again to 60s to work around multipath breakage in CA-27666 *)
		(* We then allowed it to be bumped persistently by users to work around multipath breakage in CA-28306 *)
		let t =
			if List.mem_assoc Xapi_globs.xha_timeout configuration
			then int_of_string (List.assoc Xapi_globs.xha_timeout configuration)
			else
				if List.mem_assoc Xapi_globs.default_ha_timeout other_config
				then int_of_string (List.assoc Xapi_globs.default_ha_timeout other_config)
				else int_of_float !Xapi_globs.ha_default_timeout_base in
		t
end

module Monitor = struct
	(** Control the background HA monitoring thread *)

	let request_shutdown = ref false
	let prevent_failover_actions_until = ref 0. (* protected by the request_shutdown_m too *)
	let block_delay_calls = ref false (* set to true when Pool.ha_prevent_restarts_for calls must wait *)
	let block_delay_calls_c = Condition.create () (* used to wake up all Pool.ha_prevent_restarts_for threads *)
	let m = Mutex.create ()


	(* We use this for interruptible sleeping *)
	let delay = Delay.make ()

	let thread = ref None
	let thread_m = Mutex.create ()

	let database_state_valid = ref false
	let database_state_valid_c = Condition.create ()

	(* Used to explicitly signal that we should replan *)
	let plan_out_of_date = ref true

	exception Already_started
	exception Not_started

	(** Background thread which monitors the membership set and takes action if HA is
		armed and something goes wrong *)
	let thread_body () : unit =
		Debug.name_thread "ha_monitor";
		debug "initialising HA background thread";
		(* NB we may be running this code on a slave in emergency mode *)

		Server_helpers.exec_with_new_task "HA monitor" (fun __context ->

			let statefiles = Xha_statefile.list_existing_statefiles () in

			debug "HA background thread starting";

			(* Grab the base timeout value so we can cook the reported latencies *)
			let base_t = int_of_string (Localdb.get Constants.ha_base_t) in
			let timeouts = Timeouts.derive base_t in

			(* Set up our per-host alert triggers *)

			let localhost_uuid = Helpers.get_localhost_uuid () in
			let boolean_warning alert_name alert_body_and_priority_option =
				let trigger =
					Xapi_alert.edge_trigger
						(fun old newvalue ->
							if newvalue then begin
								warn "%s" alert_name;
								begin
									match alert_body_and_priority_option with
											None -> ()
										| (Some (body, priority)) ->
											Xapi_alert.add ~name:alert_name ~priority ~cls:`Host ~obj_uuid:localhost_uuid ~body
								end
							end) in
				(* make sure we spot the case where the warning triggers immediately *)
				trigger false;
				trigger in
			(* Per-host warnings which are logged but do not generate alerts: *)
			let warning_statefile_lost = boolean_warning Api_messages.ha_statefile_lost None in
			let warning_heartbeat_approaching_timeout = boolean_warning Api_messages.ha_heartbeat_approaching_timeout None in
			let warning_statefile_approaching_timeout = boolean_warning Api_messages.ha_statefile_approaching_timeout None in
			let warning_xapi_healthcheck_approaching_timeout = boolean_warning Api_messages.ha_xapi_healthcheck_approaching_timeout None in
			(* Per-host warnings which are logged *and* generate alerts: *)
			let warning_network_bonding_error =
				boolean_warning Api_messages.ha_network_bonding_error
					(Some (Printf.sprintf "The network bond used for transmitting HA heartbeat messages on host '%s' has failed" localhost_uuid,
					Api_messages.ha_network_bonding_error_priority))  in

			(* Pool-wide warning which is logged by the master *and* generates an alert. Since this call is only ever made on the master we're
			   ok to make database calls to compute the message body without worrying about the db call blocking: *)
			let warning_all_live_nodes_lost_statefile =
				boolean_warning Api_messages.ha_statefile_lost
					(Some (Printf.sprintf "All live servers have lost access to the HA statefile",
					Api_messages.ha_statefile_lost_priority)) in

			let last_liveset_uuids = ref [] in
			let last_plan_time = ref 0. in

			(* Called on all hosts to query the liveset and update statistics + messages *)
			(* WARNING: must not touch the database or perform blocking I/O              *)
			let query_liveset_on_all_hosts () =
				let liveset =
					try
						(* XXX: if we detect the liveset has been poisoned then we're in the middle
						   of a with-statefile disable and should stop our monitor thread *)
						query_liveset ()
					with
						| Xha_error Xha_errno.Mtc_exit_daemon_is_not_present as e ->
							info "Monitor thread caught MTC_EXIT_DAEMON_IS_NOT_PRESENT; deactivating HA failover thread";
							Mutex.execute m (fun () -> request_shutdown := true);
							raise e
						| e ->
							info "Caught exception querying liveset: %s; deactivating HA failover thread"
								(ExnHelper.string_of_exn e);
							Mutex.execute m (fun () -> request_shutdown := true);
							raise e in
				debug "Liveset: %s" (Xha_interface.LiveSetInformation.to_summary_string liveset);
				(* All hosts: Feed the current latency values into the per-host RRDs (if available) *)
				Opt.iter
					(fun local ->
						Mutex.execute Xapi_ha_stats.m
							(fun () ->
								Xapi_ha_stats.enabled := true;
								(* Assume all values are ms *)
								let statefile = float_of_int (local.Xha_interface.LiveSetInformation.RawStatus.statefile_latency) /. 1000. in
								let heartbeat = float_of_int local.Xha_interface.LiveSetInformation.RawStatus.heartbeat_latency /. 1000. -. (float_of_int timeouts.Timeouts.heart_beat_interval) in
								let xapi = float_of_int (local.Xha_interface.LiveSetInformation.RawStatus.xapi_healthcheck_latency) /. 1000. in
								let t = List.map (fun vdi -> { Xapi_ha_stats.Statefile_latency.id = vdi.Static_vdis.uuid; latency = Some statefile}) statefiles in
								Xapi_ha_stats.Statefile_latency.all := t;
								Xapi_ha_stats.Heartbeat_latency.raw := Some heartbeat;
								Xapi_ha_stats.Xapi_latency.raw      := Some xapi;
							)
					) liveset.Xha_interface.LiveSetInformation.raw_status_on_local_host;

				(* All hosts: create alerts from per-host warnings (if available) *)
				debug "Processing warnings";
				Opt.iter
					(fun warning ->
						warning_statefile_lost warning.Xha_interface.LiveSetInformation.Warning.statefile_lost;
						warning_heartbeat_approaching_timeout warning.Xha_interface.LiveSetInformation.Warning.heartbeat_approaching_timeout;
						warning_statefile_approaching_timeout warning.Xha_interface.LiveSetInformation.Warning.statefile_approaching_timeout;
						warning_xapi_healthcheck_approaching_timeout warning.Xha_interface.LiveSetInformation.Warning.xapi_healthcheck_approaching_timeout;
						warning_network_bonding_error warning.Xha_interface.LiveSetInformation.Warning.network_bonding_error;
					) liveset.Xha_interface.LiveSetInformation.warning_on_local_host;
				debug "Done with warnings";
				liveset in

			(* Slaves monitor the master                                    *)
			(* WARNING: must not touch the database or perform blocking I/O *)
			let process_liveset_on_slave liveset =
				let address = Pool_role.get_master_address () in
				let master_uuid = uuid_of_host_address address in
				let master_info = Hashtbl.find liveset.Xha_interface.LiveSetInformation.hosts (Uuid.uuid_of_string master_uuid) in
				if true
					&& master_info.Xha_interface.LiveSetInformation.Host.liveness
					&& master_info.Xha_interface.LiveSetInformation.Host.master
				then debug "The node we think is the master is still alive and marked as master; this is OK"
				else begin
					warn "We think node %s (%s) is the master but the liveset disagrees" master_uuid address;
					on_master_failure ()
				end in

			(* Return the host UUIDs of all nodes in the liveset *)
			let uuids_of_liveset liveset =
				Hashtbl.fold
					(fun uuid host acc ->
						if host.Xha_interface.LiveSetInformation.Host.liveness then uuid :: acc else acc)
					liveset.Xha_interface.LiveSetInformation.hosts [] in

			(* Master performs VM restart and keeps track of the recovery plan *)
			(* WARNING: database is ok but must not perform blocking I/O       *)
			let process_liveset_on_master liveset =
				let pool = Helpers.get_pool ~__context in
				let to_tolerate = Int64.to_int (Db.Pool.get_ha_host_failures_to_tolerate ~__context ~self:pool) in
				(* let planned_for = Int64.to_int (Db.Pool.get_ha_plan_exists_for ~__context ~self:pool) in *)

				(* First consider whether VM failover actions need to happen.
				   Convert the liveset into a list of Host references used by the VM failover code *)
				let liveset_uuids = List.sort compare (uuids_of_liveset liveset) in
				if !last_liveset_uuids <> liveset_uuids then begin
					warn "Liveset looks different; assuming we need to rerun the planner";
					plan_out_of_date := true;
					last_liveset_uuids := liveset_uuids
				end;

				let liveset_refs = List.map (fun uuid -> Db.Host.get_by_uuid ~__context ~uuid:(Uuid.string_of_uuid uuid)) liveset_uuids in
				if local_failover_decisions_are_ok () then begin
					try
						Xapi_ha_vm_failover.restart_auto_run_vms ~__context liveset_refs to_tolerate
					with e ->
						error "Caught unexpected exception when executing restart plan: %s" (ExnHelper.string_of_exn e)
				end;

				(* At this point the hosts not in the liveset have been declared dead *)

				(* Next update the Host_metrics.live value to spot hosts coming back *)
				let all_hosts = Db.Host.get_all ~__context in
				let livemap = List.map (fun host -> host, List.mem host liveset_refs) all_hosts in
				List.iter (fun (host, live) ->
					Helpers.log_exn_continue
						(Printf.sprintf "updating Host_metrics.live to %b for %s" live (Ref.string_of host))
						(fun () ->
							let metrics = Db.Host.get_metrics ~__context ~self:host in
							let current = Db.Host_metrics.get_live ~__context ~self:metrics in
							let shutting_down =
								Mutex.execute Xapi_globs.hosts_which_are_shutting_down_m
									(fun () -> List.mem host !Xapi_globs.hosts_which_are_shutting_down) in
							if current <> live then begin
								(* This can only be a false -> true transient as the 'restart_auto_run_vms' function
								   has already dealt with the true -> false case. *)
								(* => live must be true. No need to consider calling current script hooks *)
								if shutting_down
								then info "Not marking host %s as live because it is shutting down" (Ref.string_of host)
								else begin
									Db.Host_metrics.set_live ~__context ~self:metrics ~value:live;
									Xapi_host_helpers.update_allowed_operations ~__context ~self:host
								end
							end
						) ()
				) livemap;

				(* Find hosts whose license expiry is in the past and forcibly disable them if necessary. *)
				let license_has_expired host =
				try
					License_check.check_expiry ~__context ~host;
					false
				with 
					| Api_errors.Server_error (code, []) when code = Api_errors.license_expired -> true
					| _ -> false (* fail safe *) in
				let expired_hosts = List.filter license_has_expired all_hosts in
				(* Find the expired ones which are still enabled *)
				let enabled_but_expired = List.filter (fun self -> Db.Host.get_enabled ~__context ~self) expired_hosts in
				List.iter 
					(fun host ->
						warn "Host uuid %s: license expired in the past; forcibly disabling" (Db.Host.get_uuid ~__context ~self:host);
						Db.Host.set_enabled ~__context ~self:host ~value:false
					) enabled_but_expired;
				(* If any host was forcibly disabled then mark the plan as out-of-date *)
				if enabled_but_expired <> [] then plan_out_of_date := true;

				(* Next update the Host.ha_statefiles and Host.ha_network_peers fields. For the network
				   peers we use whichever view is more recent: network or statefile *)
				let statefiles = Db.Pool.get_ha_statefiles ~__context ~self:pool in
				let host_host_table = List.map
					(fun host ->
						host, Hashtbl.find liveset.Xha_interface.LiveSetInformation.hosts
							(Uuid.uuid_of_string (Db.Host.get_uuid ~__context ~self:host))) all_hosts in
				List.iter (fun (host, xha_host) ->
					(* NB there is only one statefile currently *)
					let current = Db.Host.get_ha_statefiles ~__context ~self:host <> [] in
					let newval = xha_host.Xha_interface.LiveSetInformation.Host.state_file_access in
					if current <> newval
					then Db.Host.set_ha_statefiles ~__context ~self:host
						~value:(if newval then statefiles else []);
				) host_host_table;
				(* If all live hosts have lost statefile then we are running thanks to Survival Rule 2: this should be flagged to the user,
				   who should fix their storage. Note that if some hosts can see the storage but others cannot, there is no point generating
				   an alert because those who cannot are about to fence. *)
				let relying_on_rule_2 xha_host =
					true
					&& xha_host.Xha_interface.LiveSetInformation.Host.liveness              (* it is still alive *)
					&& not xha_host.Xha_interface.LiveSetInformation.Host.state_file_access (* and yet has no statefile access *)
				in
				let all_live_nodes_lost_statefile = List.fold_left (&&) true (List.map (fun (_, xha_host) -> relying_on_rule_2 xha_host) host_host_table) in
				warning_all_live_nodes_lost_statefile all_live_nodes_lost_statefile;

				(* Now the Host.ha_network_peers *)
				let subset a b = List.fold_left (fun acc x -> acc && (List.mem x b)) true a in
				let set_equals a b =
					let a' = List.setify a and b' = List.setify b in
					subset a' b' && (subset b' a') in

				(* NB raw_status_on_local_host not available if in 'Starting' state *)
				begin match liveset.Xha_interface.LiveSetInformation.raw_status_on_local_host with
					| None ->
						debug "No raw_status_on_local_host to process"
					| Some local ->
						let host_raw_table = List.map
							(fun host ->
								host, Hashtbl.find local.Xha_interface.LiveSetInformation.RawStatus.host_raw_data
									(Uuid.uuid_of_string (Db.Host.get_uuid ~__context ~self:host))) all_hosts in
						List.iter (fun (host, raw) ->
							(* Use the list of network peers given by the host recent update: statefile or network *)
							let peers =
								if raw.Xha_interface.LiveSetInformation.HostRawData.time_since_last_update_on_statefile <
									raw.Xha_interface.LiveSetInformation.HostRawData.time_since_last_heartbeat
								then raw.Xha_interface.LiveSetInformation.HostRawData.heartbeat_active_list_on_statefile
								else raw.Xha_interface.LiveSetInformation.HostRawData.heartbeat_active_list_on_heartbeat in
							let peer_strings = List.map Uuid.string_of_uuid peers in
							debug "Network peers = [%s]" (String.concat ";" peer_strings);
							let existing_strings = Db.Host.get_ha_network_peers ~__context ~self:host in
							if not(set_equals peer_strings existing_strings)
							then Db.Host.set_ha_network_peers ~__context ~self:host ~value:peer_strings)
							host_raw_table;

				end;

				let now = Unix.gettimeofday () in
				let plan_too_old = now -. !last_plan_time > !Xapi_globs.ha_monitor_plan_interval in
				if plan_too_old || !plan_out_of_date then begin
					let changed = Xapi_ha_vm_failover.update_pool_status ~__context in

					(* Extremely bad: something managed to break our careful plan *)
					if changed && not !plan_out_of_date then error "Overcommit protection failed to prevent a change which invalidated our failover plan";

					last_plan_time := now;
					plan_out_of_date := false;
				end in

			(* Wait for all hosts in the liveset to become enabled so we can start VMs on them. We wait for up to ha_monitor_startup_timeout. *)
			let wait_for_slaves_on_master () =
				(* CA-17849: need to make sure that, in the xapi startup case, all hosts have been set to dead and disabled initially *)
				info "Master HA startup waiting for explicit signal that the database state is valid";
				Mutex.execute thread_m
					(fun () ->
						while not(!database_state_valid) do
							Condition.wait database_state_valid_c thread_m
						done);

				info "Master HA startup waiting for up to %.2f for slaves in the liveset to report in and enable themselves" !Xapi_globs.ha_monitor_startup_timeout;
				let start = Unix.gettimeofday () in
				let finished = ref false in
				while Mutex.execute m (fun () -> not(!request_shutdown)) && not(!finished) do
					try
						ignore(Delay.wait delay !Xapi_globs.ha_monitor_interval);
						if Mutex.execute m (fun () -> not(!request_shutdown)) then begin
							let liveset = query_liveset_on_all_hosts () in
							let uuids = List.map Uuid.string_of_uuid (uuids_of_liveset liveset) in
							let enabled = List.map (fun uuid -> Db.Host.get_enabled ~__context ~self:(Db.Host.get_by_uuid ~__context ~uuid)) uuids in
							let enabled, disabled = List.partition (fun (_, x) -> x) (List.combine uuids enabled) in
							debug "Enabled hosts = [ %s ]; disabled hosts = [ %s ]" (String.concat "; " (List.map fst enabled)) (String.concat "; " (List.map fst disabled));

							if disabled = [] then begin
								info "Master HA startup: All live hosts are now enabled: [ %s ]" (String.concat "; " (List.map fst enabled));
								finished := true;
							end;

							if Unix.gettimeofday () -. start > !Xapi_globs.ha_monitor_startup_timeout && disabled <> [] then begin
								info "Master HA startup: Timed out waiting for all live slaves to enable themselves (have some hosts failed to attach storage?) Live but disabled hosts: [ %s ]"
									(String.concat "; " (List.map fst disabled));
								finished := true
							end;
						end;
					with e ->
						debug "Exception in HA monitor thread while waiting for slaves: %s" (ExnHelper.string_of_exn e);
						Thread.delay !Xapi_globs.ha_monitor_interval
				done in

			(* If we're the master we must wait for our live slaves to turn up before we consider restarting VMs etc *)
			if Pool_role.is_master () then wait_for_slaves_on_master ();

			(* Monitoring phase: we must assume the worst and not touch the database here *)
			while Mutex.execute m (fun () -> not(!request_shutdown)) do
				try
					ignore(Delay.wait delay !Xapi_globs.ha_monitor_interval);

					if Mutex.execute m (fun () -> not(!request_shutdown)) then begin
						let liveset = query_liveset_on_all_hosts () in
						if Pool_role.is_slave () then process_liveset_on_slave liveset;
						if Pool_role.is_master () then begin
							(* CA-23998: allow MTC to block master failover actions (ie VM restart) for a certain period of time
							   while their Level 2 VMs are being restarted. *)
							finally
								(fun () ->
									let until = Mutex.execute m
										(fun () ->
											(* Callers of 'delay' will have to wait until we have finished processing this loop iteration *)
											block_delay_calls := true;
											!prevent_failover_actions_until) in

									(* FIST *)
									while Xapi_fist.simulate_blocking_planner () do Thread.delay 1. done;

									let now = Unix.gettimeofday () in
									if now < until
									then debug "Blocking VM restart thread for at least another %.0f seconds" (until -. now)
									else process_liveset_on_master liveset)
								(fun () ->
									(* Safe to unblock callers of 'delay' now *)
									Mutex.execute m
										(fun () ->
											(* Callers of 'delay' can now safely request a delay knowing that the liveset won't be processed
											   until after the delay period. *)
											block_delay_calls := false;
											Condition.broadcast block_delay_calls_c) )
						end
					end
				with e ->
					debug "Exception in HA monitor thread: %s" (ExnHelper.string_of_exn e);
					Thread.delay !Xapi_globs.ha_monitor_interval
			done;

			debug "Re-enabling old Host_metrics.live heartbeat";
			Mutex.execute Db_gc.use_host_heartbeat_for_liveness_m
				(fun () -> Db_gc.use_host_heartbeat_for_liveness := true);

			debug "Stopping reading per-host HA stats";
			Mutex.execute Xapi_ha_stats.m
				(fun () -> Xapi_ha_stats.enabled := false);

			debug "HA background thread told to stop")

	let prevent_restarts_for seconds =
		(* Wait until the thread stops processing and is about to sleep / is already sleeping *)
		Mutex.execute m
			(fun () ->
				while !block_delay_calls = true do Condition.wait block_delay_calls_c m done;
				debug "Blocking VM restart actions for another %Ld seconds" seconds;
				prevent_failover_actions_until := Unix.gettimeofday () +. (Int64.to_float seconds);
				(* If we get a value of 0 then we immediately trigger a VM restart *)
				if seconds < 1L then Delay.signal delay
			)

	let start () =
		debug "Monitor.start()";
		debug "Disabling old heartbeat; live flag will be driven directly from xHA liveset";
		(* NB in the xapi startup case this will prevent the db_gc.single_pass from setting any live flags *)
		Mutex.execute Db_gc.use_host_heartbeat_for_liveness_m
			(fun () -> Db_gc.use_host_heartbeat_for_liveness := false);

		Mutex.execute thread_m
			(fun () ->
				match !thread with
					| Some _ -> raise Already_started
					| None ->
						(* This will cause the started thread to block until signal_database_state_valid is called *)
						request_shutdown := false;
						thread := Some (Thread.create thread_body ()))

	let signal_database_state_valid () =
		Mutex.execute thread_m
			(fun () ->
				debug "Signalling HA monitor thread that it is ok to look at the database now";
				database_state_valid := true;
				Condition.signal database_state_valid_c)

	let stop () =
		debug "Monitor.stop()";
		Mutex.execute thread_m
			(fun () ->
				match !thread with
					| None ->
						warn "Failed to stop HA monitor thread because it wasn't running. Perhaps it was stopped more than once?"
					| Some t ->
						Mutex.execute m (fun () -> request_shutdown := true; Delay.signal delay);
						Thread.join t;
						thread := None)


end

(** Called by MTC in Orlando Update 1 to temporarily block the VM restart thread. *)
let ha_prevent_restarts_for __context seconds =
	let pool = Helpers.get_pool ~__context in
	if not(Db.Pool.get_ha_enabled ~__context ~self:pool)
	then raise (Api_errors.Server_error(Api_errors.ha_not_enabled, []));
	Monitor.prevent_restarts_for seconds


(* ----------------------------- *)
(* Interaction with the redo log *)

(* This function is called when HA is enabled during run-time: flush the DB to
 * the redo-log and make future DB changes get written as deltas. *)
let redo_log_ha_enabled_during_runtime __context =
	debug "Enabling HA, so also enabling writing to redo-log";
	let pool = Helpers.get_pool ~__context in
	if Db.Pool.get_redo_log_enabled ~__context ~self:pool then begin
		info "A redo log is already in use; switch to the dedicated HA VDI.";
		Redo_log.switch ha_redo_log Xapi_globs.ha_metadata_vdi_reason
	end else begin
		info "Switching on HA redo log.";
		Redo_log.enable ha_redo_log Xapi_globs.ha_metadata_vdi_reason
			(* upon the first attempt to write a delta, it will realise that a DB flush
			 * is necessary as the I/O process will not be running *)
	end


(* This function is called when HA is disabled during run-time: stop the
 * I/O process and make future DB changes not go to the redo-log. *)
let redo_log_ha_disabled_during_runtime __context =
	debug "Disabling HA, so also disabling writing to redo-log";
	let pool = Helpers.get_pool ~__context in
	if Db.Pool.get_redo_log_enabled ~__context ~self:pool then begin
		(* switch to the other VDI *)
		info "A general redo-log is in available, independent from HA; using it now";
		Redo_log.switch ha_redo_log Xapi_globs.gen_metadata_vdi_reason
	end
	else begin
		Redo_log_usage.stop_using_redo_log ha_redo_log;
		Redo_log.disable ha_redo_log
	end

(* This function is called when HA is found to be enabled at startup, before
 * the DB backend is initialised. Read the latest DB from the block-device, and
 * make future DB changes get written as deltas. *)
let redo_log_ha_enabled_at_startup () =
	(* If we are still the master, extract any HA metadata database so we can consider population from it *)
	if Pool_role.is_master () then begin
		debug "HA is enabled, so enabling writing to redo-log";
		Redo_log.enable ha_redo_log Xapi_globs.ha_metadata_vdi_reason; (* enable the use of the redo log *)
		debug "This node is a master; attempting to extract a database from a metadata VDI";
		let db_ref = Db_backend.make () in 
		Redo_log_usage.read_from_redo_log ha_redo_log Xapi_globs.ha_metadata_db db_ref (* best effort only: does not raise any exceptions *)
	end

(* ----------------------------- *)


(** Called when xapi restarts: server may be in emergency mode at this point. We need
	to inspect the local configuration and if HA is supposed to be armed we need to
	set everything up.
	Note that
	the master shouldn't be able to activate HA while we are offline since that would cause
	us to come up with a broken configuration (the enable-HA stage has the critical task of
	synchronising the HA configuration on all the hosts). So really we only want to notice
	if the Pool has had HA disabled while we were offline. *)
let on_server_restart () =
	let armed = bool_of_string (Localdb.get Constants.ha_armed) in

	if armed then begin
		debug "HA is supposed to be armed";
		(* Make sure daemons are up *)

		let finished = ref false in
		(* Do not proceed any further until the situation is resolved.
		   XXX we might need some kind of user-override *)
		while not (!finished) do

			(* If someone has called Host.emergency_ha_disable in the background then we notice the change here *)
			if not (try bool_of_string (Localdb.get Constants.ha_armed) with _ -> false) then begin
				warn "ha_start_daemon aborted because someone has called Host.emergency_ha_disable";
				failwith "Host.emergency_ha_disable"; (* failure causes HA startup to abort *)
			end;

			try
				if Xapi_fist.ha_cannot_access_statefile () then raise (Xha_error Xha_errno.Mtc_exit_can_not_access_statefile);
				if Xapi_fist.ha_daemon_startup_failed () then failwith "simulating xha daemon startup failure";

				(* CA-21406: Try again to reattach the statefile VDI *)
				Static_vdis.reattempt_on_boot_attach ();

				let (_ : string) = call_script ha_start_daemon [] in
				finished := true;
			with
				| Xha_error Xha_errno.Mtc_exit_daemon_is_present ->
					warn "ha_start_daemon failed with MTC_EXIT_DAEMON_IS_PRESENT: continuing with startup";
					finished := true;
				| Xha_error Xha_errno.Mtc_exit_invalid_pool_state as e ->
					warn "ha_start_daemon failed with MTC_EXIT_INVALID_POOL_STATE: disabling HA on this host";
					Localdb.put Constants.ha_armed "false";
					raise e
				| Xha_error Xha_errno.Mtc_exit_can_not_access_statefile as e ->
					warn "ha_start_daemon failed with MTC_EXIT_CAN_NOT_ACCESS_STATEFILE: will contact existing master and check if HA is still enabled";

					(* check the Pool.ha_enabled on the master... assuming we can find the master. If we can't we stay here forever. *)
					let master_can_confirm_ha_is_disabled () =
						try
							let address = Pool_role.get_master_address () in
							Helpers.call_emergency_mode_functions address
								(fun rpc session_id ->
									let pool = List.hd (Client.Pool.get_all rpc session_id) in
									not (Client.Pool.get_ha_enabled rpc session_id pool)
								)
						with _ ->
							(* there's no-one for us to ask about whether HA is enabled or not *)
							false in
					if master_can_confirm_ha_is_disabled () then begin
						info "Existing master confirmed that HA is disabled pool-wide: disabling HA on this host";
						Localdb.put Constants.ha_armed "false";
						raise e
					end;
					info "Assuming HA is still enabled on the Pool and that our storage system has failed: will retry in 10s";
					Xapi_globs.slave_emergency_mode := true;
					Xapi_globs.emergency_mode_error := Api_errors.Server_error(Api_errors.ha_host_cannot_access_statefile, []);
					Helpers.touch_file !Xapi_globs.ready_file;
					Thread.delay 10.;
				| Xha_error errno ->
					error "ha_start_daemon failed with unexpected error %s: will retry in 10s" (Xha_errno.to_string errno);
					Xapi_globs.slave_emergency_mode := true;
					Xapi_globs.emergency_mode_error := Api_errors.Server_error(Api_errors.ha_heartbeat_daemon_startup_failed, []);
					Helpers.touch_file !Xapi_globs.ready_file;
					Thread.delay 10.;
				| e ->
					error "ha_start_daemon failed with unexpected exception: %s -- retrying in 10s" (ExnHelper.string_of_exn e);
					Xapi_globs.slave_emergency_mode := true;
					Xapi_globs.emergency_mode_error := Api_errors.Server_error(Api_errors.ha_heartbeat_daemon_startup_failed, []);
					Helpers.touch_file !Xapi_globs.ready_file;
					Thread.delay 10.;
		done;

		if Pool_role.is_master () then begin
			if not (local_failover_decisions_are_ok ()) then begin
				warn "ha.disable_failover_decisions flag set: not proposing myself as master";
				on_master_failure ()
			end else begin
				if propose_master ()
				then info "ha_propose_master succeeded; continuing"
				else on_master_failure ()
			end
		end;

		(* Start up the redo-log if appropriate. *)
		redo_log_ha_enabled_at_startup ();

		debug "About to start the monitor";
		Monitor.start ();
		(* We signal the monitor that the database state is valid (wrt liveness + disabledness of hosts) later *)
	end

(** Called in the master xapi startup when the database is ready. We set all hosts (including this one) to
	disabled then signal the monitor thread to look. It can then wait for slaves to turn up
	before trying to restart VMs. *)
let on_database_engine_ready () =
	info "Setting all hosts to dead and disabled. Hosts must re-enable themselves explicitly";
	Server_helpers.exec_with_new_task "Setting all hosts to dead and disabled"
		(fun __context ->
			List.iter (fun self ->
				let uuid = Db.Host.get_uuid ~__context ~self in
				let hostname = Db.Host.get_hostname ~__context ~self in
				info "Host.enabled: on_database_engine_ready: setting host %s (%s) to disabled" uuid hostname;
				Db.Host.set_enabled ~__context ~self ~value:false
			)
				(Db.Host.get_all ~__context)
		);
	Monitor.signal_database_state_valid ()

(*********************************************************************************************)
(* Internal API calls to configure individual hosts                                          *)

(** Internal API call to prevent this node making an unsafe failover decision.
	This call is idempotent. *)
let ha_disable_failover_decisions __context localhost =
	debug "Disabling failover decisions";
	(* FIST *)
	if Xapi_fist.disable_ha_disable_failover () then begin
		error "FIST: ha_disable_failover_decisions";
		failwith "FIST: ha_disable_failover_decisions"
	end;
	Localdb.put Constants.ha_disable_failover_decisions "true"

(** Internal API call to disarm localhost.
	If the daemon is missing then we return success. Either fencing was previously disabled and the
	daemon has shutdown OR the daemon has died and this node will fence shortly...
 *)
let ha_disarm_fencing __context localhost =
	try
		let (_ : string) = call_script ha_disarm_fencing [] in ()
	with Xha_error Xha_errno.Mtc_exit_daemon_is_not_present ->
		info "ha_disarm_fencing: daemon has exited so returning success"

let ha_set_excluded __context localhost =
	let (_ : string) = call_script ha_set_excluded [] in ()

(** Internal API call to stop the HA daemon.
	This call is idempotent. *)
let ha_stop_daemon __context localhost =
	Monitor.stop ();
	let (_ : string) = call_script ha_stop_daemon [] in ()

(** Emergency-mode API call to disarm localhost *)
let emergency_ha_disable __context =
	if Localdb.get Constants.ha_armed = "false"
	then raise (Api_errors.Server_error(Api_errors.ha_not_enabled, []));

	warn "Host.emergency_ha_disable: Disabling the HA subsystem on the local host only.";
	Localdb.put Constants.ha_armed "false";

	begin
		try
			ha_disarm_fencing __context ();
		with Xha_error e ->
			error "Host.emergency_ha_disable: ha_disarm_fencing failed with %s; continuing" (Xha_errno.to_string e)
	end;
	begin
		try
			ha_stop_daemon __context ();
		with Xha_error e ->
			error "Host.emergency_ha_disable: ha_stop_daemon failed with %s; continuing" (Xha_errno.to_string e)
	end;
	(* Might not be able to access the database to detach statefiles; however this isn't critical *)
	()

(** Internal API call to release any HA resources after the system has
	been shutdown.  This call is idempotent. Modified for CA-48539 to
	call vdi.deactivate before vdi.detach. *)
let ha_release_resources __context localhost =
	Monitor.stop ();

	(* Why aren't we calling Xha_statefile.detach_existing_statefiles?
	   Does Db.Pool.get_ha_statefiles return a different set of
	   statefiles than Xha_statefile.list_existing_statefiles? *)

	(* Deactivate and detach all statefile VDIs in the entire pool *)
	let statefile_vdis = Db.Pool.get_ha_statefiles ~__context ~self:(Helpers.get_pool ~__context)
	and deactiavte_and_detach_vdi vdi_str =
		let uuid = Db.VDI.get_uuid ~__context ~self:(Ref.of_string vdi_str) in
		Helpers.log_exn_continue
			(Printf.sprintf "detaching statefile VDI uuid: %s" uuid)
			(fun () ->
				Static_vdis.permanent_vdi_deactivate_by_uuid ~__context ~uuid ;
				Static_vdis.permanent_vdi_detach_by_uuid ~__context ~uuid) ()
	in List.iter deactiavte_and_detach_vdi statefile_vdis ;

	(* Deactivate and detach any metadata VDIs *)
	Helpers.log_exn_continue
		(Printf.sprintf "deactivating and detaching metadata VDIs")
		(fun () -> Xha_metadata_vdi.deactivate_and_detach_existing ~__context) ();

	(* At this point a restart won't enable the HA subsystem *)
	Localdb.put Constants.ha_armed "false"

(** Internal API call which blocks until this node's xHA daemon spots the invalid statefile
	and exits cleanly. If the daemon survives but the statefile access is lost then this function
	will return an exception and the no-statefile shutdown can be attempted.
 *)
let ha_wait_for_shutdown_via_statefile __context localhost =
	try
		while true do
			let liveset = query_liveset () in

			let hosts = liveset.Xha_interface.LiveSetInformation.hosts in
			let me = Hashtbl.find hosts liveset.Xha_interface.LiveSetInformation.local_host_id in
			(* If we have no statefile access or if it looks corrupted, fail the operation *)
			if false
				|| not me.Xha_interface.LiveSetInformation.Host.state_file_access
				|| me.Xha_interface.LiveSetInformation.Host.state_file_corrupted
			then raise (Api_errors.Server_error(Api_errors.ha_lost_statefile, []));

			(* check to see if this node still has statefile access *)
			Thread.delay 5.
		done
	with
		| Xha_error Xha_errno.Mtc_exit_daemon_is_not_present ->
			info "ha_wait_for_shutdown_via_statefile: daemon has exited so returning success"


(** Attach the statefile VDIs and return the resulting list of paths in dom0 *)
let attach_statefiles ~__context statevdis =
	(* First GC any existing statefiles: these are not needed any more *)
	info "Detaching any existing statefiles: these are not needed any more";
	Xha_statefile.detach_existing_statefiles ~__context;

	let paths = ref [] in
	begin
		try
			List.iter
				(fun vdi ->
					info "Attempting to permanently attach statefile VDI: %s" (Ref.string_of vdi);
					paths := Static_vdis.permanent_vdi_attach ~__context ~vdi ~reason:Xha_statefile.reason:: !paths) statevdis
		with e ->
			error "Caught exception attaching statefile: %s" (ExnHelper.string_of_exn e);
			List.iter
				(fun vdi ->
					Helpers.log_exn_continue
						(Printf.sprintf "detaching statefile: %s" (Ref.string_of vdi))
						(fun () -> Static_vdis.permanent_vdi_detach ~__context ~vdi) ()
				) statevdis;
			raise e
	end;
	!paths

(** Attach the metadata VDI and return the resulting path in dom0 *)
let attach_metadata_vdi ~__context vdi =
	info "Detaching any existing metadata volume: these are not needed anymore";
	Xha_metadata_vdi.detach_existing ~__context;<commit_msg>CA-71770: More user friendly error message when enabling HA with an SR that does not exist.<commit_after>(*
 * Copyright (C) 2006-2009 Citrix Systems Inc.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation; version 2.1 only. with the special
 * exception on linking described in file LICENSE.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *)
(** Functions for implementing 'High Availability' (HA). File is divided into 3 sections:
	+ scripts and functions which form part of the HA subsystem interface
	+ internal API calls used for arming and disarming individual hosts
	+ external API calls (Pool.enable_ha, Pool.disable_ha) used for turning on/off HA pool-wide
	* @group High Availability (HA)
 *)

module D = Debug.Debugger(struct let name="xapi_ha" end)
open D

open Listext
open Stringext
open Threadext
open Pervasiveext
open Forkhelpers
open Client
open Threadext
open Db_filter_types

(* Create a redo_log instance to use for HA. *)
let ha_redo_log = Redo_log.create ~name:"HA redo log" ~state_change_callback:None ~read_only:false

(*********************************************************************************************)
(* Interface with the low-level HA subsystem                                                 *)

let ha_set_pool_state = Filename.concat Fhs.xhadir "ha_set_pool_state"
let ha_start_daemon = Filename.concat Fhs.xhadir "ha_start_daemon"
let ha_stop_daemon = Filename.concat Fhs.xhadir "ha_stop_daemon"
let ha_query_liveset = Filename.concat Fhs.xhadir "ha_query_liveset"
let ha_propose_master = Filename.concat Fhs.xhadir "ha_propose_master"
let ha_disarm_fencing = Filename.concat Fhs.xhadir "ha_disarm_fencing"
let ha_set_excluded = Filename.concat Fhs.xhadir "ha_set_excluded"
let fence_path = Filename.concat Fhs.libexecdir "fence"
(* Unused: let ha_clear_excluded = Filename.concat Fhs.xhadir "ha_clear_excluded" *)

(** The xHA scripts throw these exceptions: *)
exception Xha_error of Xha_errno.code

(** Only call one xHA script at a time *)
let ha_script_m = Mutex.create ()

let call_script ?log_successful_output script args =
	try
		Mutex.execute ha_script_m (fun () -> Helpers.call_script ?log_successful_output script args)
	with Forkhelpers.Spawn_internal_error(stderr, stdout, Unix.WEXITED n) ->
		let code = Xha_errno.of_int n in
		warn "%s %s returned %s (%s)" script (String.concat " " args)
			(Xha_errno.to_string code) (Xha_errno.to_description_string code);
		raise (Xha_error code)

(** Returns the current live set info *)
let query_liveset() =
	let txt = call_script ~log_successful_output:false ha_query_liveset [] in
	Xha_interface.LiveSetInformation.of_xml_string txt

(** Returns true if this node has statefile access *)
let i_have_statefile_access () =
	try
		let liveset = query_liveset () in
		let hosts = liveset.Xha_interface.LiveSetInformation.hosts in
		let me = Hashtbl.find hosts liveset.Xha_interface.LiveSetInformation.local_host_id in
		me.Xha_interface.LiveSetInformation.Host.state_file_access && not (me.Xha_interface.LiveSetInformation.Host.state_file_corrupted)
	with e ->
		info "Caught exception querying liveset; assuming we have no statefile access: %s" (ExnHelper.string_of_exn e);
		false

(** Returns true if this node is allowed to be the master *)
let propose_master () =
	try
		let result = call_script ha_propose_master [] in
		String.rtrim result = "TRUE"
	with Xha_error e ->
		error "ha_propose_master threw unexpected exception: %s" (Xha_errno.to_string e);
		false

(** Returns true if local failover decisions have not been disabled on this node *)
let local_failover_decisions_are_ok () =
	try not(bool_of_string (Localdb.get Constants.ha_disable_failover_decisions)) with _ -> true

(** Since the liveset info doesn't include the host IP address, we persist these ourselves *)
let write_uuid_to_ip_mapping ~__context =
	let table = List.map (fun (_, host) -> host.API.host_uuid, host.API.host_address) (Db.Host.get_all_records ~__context) in
	let v = String_marshall_helper.map (fun x -> x) (fun x -> x) table in
	Localdb.put Constants.ha_peers v

(** Since the liveset info doesn't include the host IP address, we persist these ourselves *)
let get_uuid_to_ip_mapping () =
	let v = Localdb.get Constants.ha_peers in
	String_unmarshall_helper.map (fun x -> x) (fun x -> x) v

(** Without using the Pool's database, returns the IP address of a particular host
	named by UUID. *)
let address_of_host_uuid uuid =
	let table = get_uuid_to_ip_mapping () in
	if not(List.mem_assoc uuid table) then begin
		error "Failed to find the IP address of host UUID %s" uuid;
		raise Not_found
	end else List.assoc uuid table

(** Without using the Pool's database, returns the UUID of a particular host named by
	heartbeat IP address. This is only necesary because the liveset info doesn't include
	the host IP address *)
let uuid_of_host_address address =
	let table = List.map (fun (k, v) -> v, k) (get_uuid_to_ip_mapping ()) in
	if not(List.mem_assoc address table) then begin
		error "Failed to find the UUID address of host with address %s" address;
		raise Not_found
	end else List.assoc address table

(** Called in two circumstances:
	1. When I started up I thought I was the master but my proposal was rejected by the
	heartbeat component.
	2. I was happily running as someone's slave but they left the liveset.
 *)
let on_master_failure () =
	(* The plan is: keep asking if I should be the master. If I'm rejected then query the
	   live set and see if someone else has been marked as master, if so become a slave of them. *)

	let become_master () =
		info "This node will become the master";
		Xapi_pool_transition.become_master ();
		info "Waiting for server restart";
		while true do Thread.delay 3600. done in
	let become_slave_of uuid =
		let address = address_of_host_uuid uuid in
		info "This node will become the slave of host %s (%s)" uuid address;
		Xapi_pool_transition.become_another_masters_slave address;
		(* XXX CA-16388: prevent blocking *)
		Thread.delay 15.;
		error "Failed to flush and exit properly; forcibly exiting";
		exit Xapi_globs.restart_return_code in

	let finished = ref false in
	while not !finished do
		(* When HA is disabled without the statefile we set a flag to indicate that this node
		   cannot transition to master automatically on boot. This is to prevent failures during
		   the 'disarm fencing' step which cause some nodes to not fence themselves when they should. *)
		if local_failover_decisions_are_ok () && propose_master () then begin
			info "ha_propose_master succeeded";
			become_master ();
			finished := true
		end else begin
			if local_failover_decisions_are_ok ()
			then info "ha_propose_master failed: looking for another master"
			else info "ha_can_not_be_master_on_next_boot set: I cannot be master; looking for another master";

			let liveset = query_liveset () in
			match Hashtbl.fold
				(fun uuid host acc ->
					if host.Xha_interface.LiveSetInformation.Host.master
						&& host.Xha_interface.LiveSetInformation.Host.liveness (* CP-25481: a dead host may still have the master lock *)
					then uuid :: acc else acc) liveset.Xha_interface.LiveSetInformation.hosts [] with
					| [] ->
						info "no other master exists yet; waiting 5 seconds and retrying";
						Thread.delay 5.
					| uuid :: [] ->
						become_slave_of (Uuid.string_of_uuid uuid)
					| xs ->
						(* should never happen *)
						error "multiple masters reported: [ %s ]; failing"
							(String.concat "; " (List.map Uuid.string_of_uuid xs));
						failwith "multiple masters"
		end
	done

module Timeouts = struct
	type t = {
		heart_beat_interval: int;
		state_file_interval: int;
		heart_beat_timeout: int;
		state_file_timeout: int;
		heart_beat_watchdog_timeout: int;
		state_file_watchdog_timeout: int;
		boot_join_timeout: int;
		enable_join_timeout: int;
		xapi_healthcheck_timeout: int;
		xapi_healthcheck_interval: int;
		xapi_restart_timeout: int;
		xapi_restart_attempts: int;
	}
	let derive (t: int) =
		(* xHA interface section 4.1.4.1.1 Formula of key timeouts *)
		(* t >= 10 *)
		if t < 10 then failwith "constraint violation: timeout >= 10";
		(* All other values are derived from this single parameter *)
		let heart_beat_interval = (t + 10) / 10 in
		let state_file_interval = heart_beat_interval in
		let heart_beat_timeout = t in
		let state_file_timeout = t in
		let heart_beat_watchdog_timeout = t in
		let state_file_watchdog_timeout = t + 15 in
		let boot_join_timeout = t + 60 in
		let enable_join_timeout = boot_join_timeout in

		{ heart_beat_interval = heart_beat_interval;
		state_file_interval = state_file_interval;
		heart_beat_timeout = heart_beat_timeout;
		state_file_timeout = state_file_timeout;
		heart_beat_watchdog_timeout = heart_beat_watchdog_timeout;
		state_file_watchdog_timeout = state_file_watchdog_timeout;
		boot_join_timeout = boot_join_timeout;
		enable_join_timeout = enable_join_timeout;

		xapi_healthcheck_interval = !Xapi_globs.ha_xapi_healthcheck_interval;
		xapi_healthcheck_timeout = !Xapi_globs.ha_xapi_healthcheck_timeout;
		xapi_restart_attempts = !Xapi_globs.ha_xapi_restart_attempts;
		xapi_restart_timeout = !Xapi_globs.ha_xapi_restart_timeout; (* 180s is max start delay and 60s max shutdown delay in the initscript *)
		}

	(** Returns the base timeout value from which the rest are derived *)
	let get_base_t ~__context =
		let pool = Helpers.get_pool ~__context in
		let other_config = Db.Pool.get_other_config ~__context ~self:pool in
		let configuration = Db.Pool.get_ha_configuration ~__context ~self:pool in
		(* xHA built-in default is 30s. We've bumped ours to 40s to make TC7710/CA-17639 happier *)
		(* We then bumped it again to 60s to work around multipath breakage in CA-27666 *)
		(* We then allowed it to be bumped persistently by users to work around multipath breakage in CA-28306 *)
		let t =
			if List.mem_assoc Xapi_globs.xha_timeout configuration
			then int_of_string (List.assoc Xapi_globs.xha_timeout configuration)
			else
				if List.mem_assoc Xapi_globs.default_ha_timeout other_config
				then int_of_string (List.assoc Xapi_globs.default_ha_timeout other_config)
				else int_of_float !Xapi_globs.ha_default_timeout_base in
		t
end

module Monitor = struct
	(** Control the background HA monitoring thread *)

	let request_shutdown = ref false
	let prevent_failover_actions_until = ref 0. (* protected by the request_shutdown_m too *)
	let block_delay_calls = ref false (* set to true when Pool.ha_prevent_restarts_for calls must wait *)
	let block_delay_calls_c = Condition.create () (* used to wake up all Pool.ha_prevent_restarts_for threads *)
	let m = Mutex.create ()


	(* We use this for interruptible sleeping *)
	let delay = Delay.make ()

	let thread = ref None
	let thread_m = Mutex.create ()

	let database_state_valid = ref false
	let database_state_valid_c = Condition.create ()

	(* Used to explicitly signal that we should replan *)
	let plan_out_of_date = ref true

	exception Already_started
	exception Not_started

	(** Background thread which monitors the membership set and takes action if HA is
		armed and something goes wrong *)
	let thread_body () : unit =
		Debug.name_thread "ha_monitor";
		debug "initialising HA background thread";
		(* NB we may be running this code on a slave in emergency mode *)

		Server_helpers.exec_with_new_task "HA monitor" (fun __context ->

			let statefiles = Xha_statefile.list_existing_statefiles () in

			debug "HA background thread starting";

			(* Grab the base timeout value so we can cook the reported latencies *)
			let base_t = int_of_string (Localdb.get Constants.ha_base_t) in
			let timeouts = Timeouts.derive base_t in

			(* Set up our per-host alert triggers *)

			let localhost_uuid = Helpers.get_localhost_uuid () in
			let boolean_warning alert_name alert_body_and_priority_option =
				let trigger =
					Xapi_alert.edge_trigger
						(fun old newvalue ->
							if newvalue then begin
								warn "%s" alert_name;
								begin
									match alert_body_and_priority_option with
											None -> ()
										| (Some (body, priority)) ->
											Xapi_alert.add ~name:alert_name ~priority ~cls:`Host ~obj_uuid:localhost_uuid ~body
								end
							end) in
				(* make sure we spot the case where the warning triggers immediately *)
				trigger false;
				trigger in
			(* Per-host warnings which are logged but do not generate alerts: *)
			let warning_statefile_lost = boolean_warning Api_messages.ha_statefile_lost None in
			let warning_heartbeat_approaching_timeout = boolean_warning Api_messages.ha_heartbeat_approaching_timeout None in
			let warning_statefile_approaching_timeout = boolean_warning Api_messages.ha_statefile_approaching_timeout None in
			let warning_xapi_healthcheck_approaching_timeout = boolean_warning Api_messages.ha_xapi_healthcheck_approaching_timeout None in
			(* Per-host warnings which are logged *and* generate alerts: *)
			let warning_network_bonding_error =
				boolean_warning Api_messages.ha_network_bonding_error
					(Some (Printf.sprintf "The network bond used for transmitting HA heartbeat messages on host '%s' has failed" localhost_uuid,
					Api_messages.ha_network_bonding_error_priority))  in

			(* Pool-wide warning which is logged by the master *and* generates an alert. Since this call is only ever made on the master we're
			   ok to make database calls to compute the message body without worrying about the db call blocking: *)
			let warning_all_live_nodes_lost_statefile =
				boolean_warning Api_messages.ha_statefile_lost
					(Some (Printf.sprintf "All live servers have lost access to the HA statefile",
					Api_messages.ha_statefile_lost_priority)) in

			let last_liveset_uuids = ref [] in
			let last_plan_time = ref 0. in

			(* Called on all hosts to query the liveset and update statistics + messages *)
			(* WARNING: must not touch the database or perform blocking I/O              *)
			let query_liveset_on_all_hosts () =
				let liveset =
					try
						(* XXX: if we detect the liveset has been poisoned then we're in the middle
						   of a with-statefile disable and should stop our monitor thread *)
						query_liveset ()
					with
						| Xha_error Xha_errno.Mtc_exit_daemon_is_not_present as e ->
							info "Monitor thread caught MTC_EXIT_DAEMON_IS_NOT_PRESENT; deactivating HA failover thread";
							Mutex.execute m (fun () -> request_shutdown := true);
							raise e
						| e ->
							info "Caught exception querying liveset: %s; deactivating HA failover thread"
								(ExnHelper.string_of_exn e);
							Mutex.execute m (fun () -> request_shutdown := true);
							raise e in
				debug "Liveset: %s" (Xha_interface.LiveSetInformation.to_summary_string liveset);
				(* All hosts: Feed the current latency values into the per-host RRDs (if available) *)
				Opt.iter
					(fun local ->
						Mutex.execute Xapi_ha_stats.m
							(fun () ->
								Xapi_ha_stats.enabled := true;
								(* Assume all values are ms *)
								let statefile = float_of_int (local.Xha_interface.LiveSetInformation.RawStatus.statefile_latency) /. 1000. in
								let heartbeat = float_of_int local.Xha_interface.LiveSetInformation.RawStatus.heartbeat_latency /. 1000. -. (float_of_int timeouts.Timeouts.heart_beat_interval) in
								let xapi = float_of_int (local.Xha_interface.LiveSetInformation.RawStatus.xapi_healthcheck_latency) /. 1000. in
								let t = List.map (fun vdi -> { Xapi_ha_stats.Statefile_latency.id = vdi.Static_vdis.uuid; latency = Some statefile}) statefiles in
								Xapi_ha_stats.Statefile_latency.all := t;
								Xapi_ha_stats.Heartbeat_latency.raw := Some heartbeat;
								Xapi_ha_stats.Xapi_latency.raw      := Some xapi;
							)
					) liveset.Xha_interface.LiveSetInformation.raw_status_on_local_host;

				(* All hosts: create alerts from per-host warnings (if available) *)
				debug "Processing warnings";
				Opt.iter
					(fun warning ->
						warning_statefile_lost warning.Xha_interface.LiveSetInformation.Warning.statefile_lost;
						warning_heartbeat_approaching_timeout warning.Xha_interface.LiveSetInformation.Warning.heartbeat_approaching_timeout;
						warning_statefile_approaching_timeout warning.Xha_interface.LiveSetInformation.Warning.statefile_approaching_timeout;
						warning_xapi_healthcheck_approaching_timeout warning.Xha_interface.LiveSetInformation.Warning.xapi_healthcheck_approaching_timeout;
						warning_network_bonding_error warning.Xha_interface.LiveSetInformation.Warning.network_bonding_error;
					) liveset.Xha_interface.LiveSetInformation.warning_on_local_host;
				debug "Done with warnings";
				liveset in

			(* Slaves monitor the master                                    *)
			(* WARNING: must not touch the database or perform blocking I/O *)
			let process_liveset_on_slave liveset =
				let address = Pool_role.get_master_address () in
				let master_uuid = uuid_of_host_address address in
				let master_info = Hashtbl.find liveset.Xha_interface.LiveSetInformation.hosts (Uuid.uuid_of_string master_uuid) in
				if true
					&& master_info.Xha_interface.LiveSetInformation.Host.liveness
					&& master_info.Xha_interface.LiveSetInformation.Host.master
				then debug "The node we think is the master is still alive and marked as master; this is OK"
				else begin
					warn "We think node %s (%s) is the master but the liveset disagrees" master_uuid address;
					on_master_failure ()
				end in

			(* Return the host UUIDs of all nodes in the liveset *)
			let uuids_of_liveset liveset =
				Hashtbl.fold
					(fun uuid host acc ->
						if host.Xha_interface.LiveSetInformation.Host.liveness then uuid :: acc else acc)
					liveset.Xha_interface.LiveSetInformation.hosts [] in

			(* Master performs VM restart and keeps track of the recovery plan *)
			(* WARNING: database is ok but must not perform blocking I/O       *)
			let process_liveset_on_master liveset =
				let pool = Helpers.get_pool ~__context in
				let to_tolerate = Int64.to_int (Db.Pool.get_ha_host_failures_to_tolerate ~__context ~self:pool) in
				(* let planned_for = Int64.to_int (Db.Pool.get_ha_plan_exists_for ~__context ~self:pool) in *)

				(* First consider whether VM failover actions need to happen.
				   Convert the liveset into a list of Host references used by the VM failover code *)
				let liveset_uuids = List.sort compare (uuids_of_liveset liveset) in
				if !last_liveset_uuids <> liveset_uuids then begin
					warn "Liveset looks different; assuming we need to rerun the planner";
					plan_out_of_date := true;
					last_liveset_uuids := liveset_uuids
				end;

				let liveset_refs = List.map (fun uuid -> Db.Host.get_by_uuid ~__context ~uuid:(Uuid.string_of_uuid uuid)) liveset_uuids in
				if local_failover_decisions_are_ok () then begin
					try
						Xapi_ha_vm_failover.restart_auto_run_vms ~__context liveset_refs to_tolerate
					with e ->
						error "Caught unexpected exception when executing restart plan: %s" (ExnHelper.string_of_exn e)
				end;

				(* At this point the hosts not in the liveset have been declared dead *)

				(* Next update the Host_metrics.live value to spot hosts coming back *)
				let all_hosts = Db.Host.get_all ~__context in
				let livemap = List.map (fun host -> host, List.mem host liveset_refs) all_hosts in
				List.iter (fun (host, live) ->
					Helpers.log_exn_continue
						(Printf.sprintf "updating Host_metrics.live to %b for %s" live (Ref.string_of host))
						(fun () ->
							let metrics = Db.Host.get_metrics ~__context ~self:host in
							let current = Db.Host_metrics.get_live ~__context ~self:metrics in
							let shutting_down =
								Mutex.execute Xapi_globs.hosts_which_are_shutting_down_m
									(fun () -> List.mem host !Xapi_globs.hosts_which_are_shutting_down) in
							if current <> live then begin
								(* This can only be a false -> true transient as the 'restart_auto_run_vms' function
								   has already dealt with the true -> false case. *)
								(* => live must be true. No need to consider calling current script hooks *)
								if shutting_down
								then info "Not marking host %s as live because it is shutting down" (Ref.string_of host)
								else begin
									Db.Host_metrics.set_live ~__context ~self:metrics ~value:live;
									Xapi_host_helpers.update_allowed_operations ~__context ~self:host
								end
							end
						) ()
				) livemap;

				(* Find hosts whose license expiry is in the past and forcibly disable them if necessary. *)
				let license_has_expired host =
				try
					License_check.check_expiry ~__context ~host;
					false
				with 
					| Api_errors.Server_error (code, []) when code = Api_errors.license_expired -> true
					| _ -> false (* fail safe *) in
				let expired_hosts = List.filter license_has_expired all_hosts in
				(* Find the expired ones which are still enabled *)
				let enabled_but_expired = List.filter (fun self -> Db.Host.get_enabled ~__context ~self) expired_hosts in
				List.iter 
					(fun host ->
						warn "Host uuid %s: license expired in the past; forcibly disabling" (Db.Host.get_uuid ~__context ~self:host);
						Db.Host.set_enabled ~__context ~self:host ~value:false
					) enabled_but_expired;
				(* If any host was forcibly disabled then mark the plan as out-of-date *)
				if enabled_but_expired <> [] then plan_out_of_date := true;

				(* Next update the Host.ha_statefiles and Host.ha_network_peers fields. For the network
				   peers we use whichever view is more recent: network or statefile *)
				let statefiles = Db.Pool.get_ha_statefiles ~__context ~self:pool in
				let host_host_table = List.map
					(fun host ->
						host, Hashtbl.find liveset.Xha_interface.LiveSetInformation.hosts
							(Uuid.uuid_of_string (Db.Host.get_uuid ~__context ~self:host))) all_hosts in
				List.iter (fun (host, xha_host) ->
					(* NB there is only one statefile currently *)
					let current = Db.Host.get_ha_statefiles ~__context ~self:host <> [] in
					let newval = xha_host.Xha_interface.LiveSetInformation.Host.state_file_access in
					if current <> newval
					then Db.Host.set_ha_statefiles ~__context ~self:host
						~value:(if newval then statefiles else []);
				) host_host_table;
				(* If all live hosts have lost statefile then we are running thanks to Survival Rule 2: this should be flagged to the user,
				   who should fix their storage. Note that if some hosts can see the storage but others cannot, there is no point generating
				   an alert because those who cannot are about to fence. *)
				let relying_on_rule_2 xha_host =
					true
					&& xha_host.Xha_interface.LiveSetInformation.Host.liveness              (* it is still alive *)
					&& not xha_host.Xha_interface.LiveSetInformation.Host.state_file_access (* and yet has no statefile access *)
				in
				let all_live_nodes_lost_statefile = List.fold_left (&&) true (List.map (fun (_, xha_host) -> relying_on_rule_2 xha_host) host_host_table) in
				warning_all_live_nodes_lost_statefile all_live_nodes_lost_statefile;

				(* Now the Host.ha_network_peers *)
				let subset a b = List.fold_left (fun acc x -> acc && (List.mem x b)) true a in
				let set_equals a b =
					let a' = List.setify a and b' = List.setify b in
					subset a' b' && (subset b' a') in

				(* NB raw_status_on_local_host not available if in 'Starting' state *)
				begin match liveset.Xha_interface.LiveSetInformation.raw_status_on_local_host with
					| None ->
						debug "No raw_status_on_local_host to process"
					| Some local ->
						let host_raw_table = List.map
							(fun host ->
								host, Hashtbl.find local.Xha_interface.LiveSetInformation.RawStatus.host_raw_data
									(Uuid.uuid_of_string (Db.Host.get_uuid ~__context ~self:host))) all_hosts in
						List.iter (fun (host, raw) ->
							(* Use the list of network peers given by the host recent update: statefile or network *)
							let peers =
								if raw.Xha_interface.LiveSetInformation.HostRawData.time_since_last_update_on_statefile <
									raw.Xha_interface.LiveSetInformation.HostRawData.time_since_last_heartbeat
								then raw.Xha_interface.LiveSetInformation.HostRawData.heartbeat_active_list_on_statefile
								else raw.Xha_interface.LiveSetInformation.HostRawData.heartbeat_active_list_on_heartbeat in
							let peer_strings = List.map Uuid.string_of_uuid peers in
							debug "Network peers = [%s]" (String.concat ";" peer_strings);
							let existing_strings = Db.Host.get_ha_network_peers ~__context ~self:host in
							if not(set_equals peer_strings existing_strings)
							then Db.Host.set_ha_network_peers ~__context ~self:host ~value:peer_strings)
							host_raw_table;

				end;

				let now = Unix.gettimeofday () in
				let plan_too_old = now -. !last_plan_time > !Xapi_globs.ha_monitor_plan_interval in
				if plan_too_old || !plan_out_of_date then begin
					let changed = Xapi_ha_vm_failover.update_pool_status ~__context in

					(* Extremely bad: something managed to break our careful plan *)
					if changed && not !plan_out_of_date then error "Overcommit protection failed to prevent a change which invalidated our failover plan";

					last_plan_time := now;
					plan_out_of_date := false;
				end in

			(* Wait for all hosts in the liveset to become enabled so we can start VMs on them. We wait for up to ha_monitor_startup_timeout. *)
			let wait_for_slaves_on_master () =
				(* CA-17849: need to make sure that, in the xapi startup case, all hosts have been set to dead and disabled initially *)
				info "Master HA startup waiting for explicit signal that the database state is valid";
				Mutex.execute thread_m
					(fun () ->
						while not(!database_state_valid) do
							Condition.wait database_state_valid_c thread_m
						done);

				info "Master HA startup waiting for up to %.2f for slaves in the liveset to report in and enable themselves" !Xapi_globs.ha_monitor_startup_timeout;
				let start = Unix.gettimeofday () in
				let finished = ref false in
				while Mutex.execute m (fun () -> not(!request_shutdown)) && not(!finished) do
					try
						ignore(Delay.wait delay !Xapi_globs.ha_monitor_interval);
						if Mutex.execute m (fun () -> not(!request_shutdown)) then begin
							let liveset = query_liveset_on_all_hosts () in
							let uuids = List.map Uuid.string_of_uuid (uuids_of_liveset liveset) in
							let enabled = List.map (fun uuid -> Db.Host.get_enabled ~__context ~self:(Db.Host.get_by_uuid ~__context ~uuid)) uuids in
							let enabled, disabled = List.partition (fun (_, x) -> x) (List.combine uuids enabled) in
							debug "Enabled hosts = [ %s ]; disabled hosts = [ %s ]" (String.concat "; " (List.map fst enabled)) (String.concat "; " (List.map fst disabled));

							if disabled = [] then begin
								info "Master HA startup: All live hosts are now enabled: [ %s ]" (String.concat "; " (List.map fst enabled));
								finished := true;
							end;

							if Unix.gettimeofday () -. start > !Xapi_globs.ha_monitor_startup_timeout && disabled <> [] then begin
								info "Master HA startup: Timed out waiting for all live slaves to enable themselves (have some hosts failed to attach storage?) Live but disabled hosts: [ %s ]"
									(String.concat "; " (List.map fst disabled));
								finished := true
							end;
						end;
					with e ->
						debug "Exception in HA monitor thread while waiting for slaves: %s" (ExnHelper.string_of_exn e);
						Thread.delay !Xapi_globs.ha_monitor_interval
				done in

			(* If we're the master we must wait for our live slaves to turn up before we consider restarting VMs etc *)
			if Pool_role.is_master () then wait_for_slaves_on_master ();

			(* Monitoring phase: we must assume the worst and not touch the database here *)
			while Mutex.execute m (fun () -> not(!request_shutdown)) do
				try
					ignore(Delay.wait delay !Xapi_globs.ha_monitor_interval);

					if Mutex.execute m (fun () -> not(!request_shutdown)) then begin
						let liveset = query_liveset_on_all_hosts () in
						if Pool_role.is_slave () then process_liveset_on_slave liveset;
						if Pool_role.is_master () then begin
							(* CA-23998: allow MTC to block master failover actions (ie VM restart) for a certain period of time
							   while their Level 2 VMs are being restarted. *)
							finally
								(fun () ->
									let until = Mutex.execute m
										(fun () ->
											(* Callers of 'delay' will have to wait until we have finished processing this loop iteration *)
											block_delay_calls := true;
											!prevent_failover_actions_until) in

									(* FIST *)
									while Xapi_fist.simulate_blocking_planner () do Thread.delay 1. done;

									let now = Unix.gettimeofday () in
									if now < until
									then debug "Blocking VM restart thread for at least another %.0f seconds" (until -. now)
									else process_liveset_on_master liveset)
								(fun () ->
									(* Safe to unblock callers of 'delay' now *)
									Mutex.execute m
										(fun () ->
											(* Callers of 'delay' can now safely request a delay knowing that the liveset won't be processed
											   until after the delay period. *)
											block_delay_calls := false;
											Condition.broadcast block_delay_calls_c) )
						end
					end
				with e ->
					debug "Exception in HA monitor thread: %s" (ExnHelper.string_of_exn e);
					Thread.delay !Xapi_globs.ha_monitor_interval
			done;

			debug "Re-enabling old Host_metrics.live heartbeat";
			Mutex.execute Db_gc.use_host_heartbeat_for_liveness_m
				(fun () -> Db_gc.use_host_heartbeat_for_liveness := true);

			debug "Stopping reading per-host HA stats";
			Mutex.execute Xapi_ha_stats.m
				(fun () -> Xapi_ha_stats.enabled := false);

			debug "HA background thread told to stop")

	let prevent_restarts_for seconds =
		(* Wait until the thread stops processing and is about to sleep / is already sleeping *)
		Mutex.execute m
			(fun () ->
				while !block_delay_calls = true do Condition.wait block_delay_calls_c m done;
				debug "Blocking VM restart actions for another %Ld seconds" seconds;
				prevent_failover_actions_until := Unix.gettimeofday () +. (Int64.to_float seconds);
				(* If we get a value of 0 then we immediately trigger a VM restart *)
				if seconds < 1L then Delay.signal delay
			)

	let start () =
		debug "Monitor.start()";
		debug "Disabling old heartbeat; live flag will be driven directly from xHA liveset";
		(* NB in the xapi startup case this will prevent the db_gc.single_pass from setting any live flags *)
		Mutex.execute Db_gc.use_host_heartbeat_for_liveness_m
			(fun () -> Db_gc.use_host_heartbeat_for_liveness := false);

		Mutex.execute thread_m
			(fun () ->
				match !thread with
					| Some _ -> raise Already_started
					| None ->
						(* This will cause the started thread to block until signal_database_state_valid is called *)
						request_shutdown := false;
						thread := Some (Thread.create thread_body ()))

	let signal_database_state_valid () =
		Mutex.execute thread_m
			(fun () ->
				debug "Signalling HA monitor thread that it is ok to look at the database now";
				database_state_valid := true;
				Condition.signal database_state_valid_c)

	let stop () =
		debug "Monitor.stop()";
		Mutex.execute thread_m
			(fun () ->
				match !thread with
					| None ->
						warn "Failed to stop HA monitor thread because it wasn't running. Perhaps it was stopped more than once?"
					| Some t ->
						Mutex.execute m (fun () -> request_shutdown := true; Delay.signal delay);
						Thread.join t;
						thread := None)


end

(** Called by MTC in Orlando Update 1 to temporarily block the VM restart thread. *)
let ha_prevent_restarts_for __context seconds =
	let pool = Helpers.get_pool ~__context in
	if not(Db.Pool.get_ha_enabled ~__context ~self:pool)
	then raise (Api_errors.Server_error(Api_errors.ha_not_enabled, []));
	Monitor.prevent_restarts_for seconds


(* ----------------------------- *)
(* Interaction with the redo log *)

(* This function is called when HA is enabled during run-time: flush the DB to
 * the redo-log and make future DB changes get written as deltas. *)
let redo_log_ha_enabled_during_runtime __context =
	debug "Enabling HA, so also enabling writing to redo-log";
	let pool = Helpers.get_pool ~__context in
	if Db.Pool.get_redo_log_enabled ~__context ~self:pool then begin
		info "A redo log is already in use; switch to the dedicated HA VDI.";
		Redo_log.switch ha_redo_log Xapi_globs.ha_metadata_vdi_reason
	end else begin
		info "Switching on HA redo log.";
		Redo_log.enable ha_redo_log Xapi_globs.ha_metadata_vdi_reason
			(* upon the first attempt to write a delta, it will realise that a DB flush
			 * is necessary as the I/O process will not be running *)
	end


(* This function is called when HA is disabled during run-time: stop the
 * I/O process and make future DB changes not go to the redo-log. *)
let redo_log_ha_disabled_during_runtime __context =
	debug "Disabling HA, so also disabling writing to redo-log";
	let pool = Helpers.get_pool ~__context in
	if Db.Pool.get_redo_log_enabled ~__context ~self:pool then begin
		(* switch to the other VDI *)
		info "A general redo-log is in available, independent from HA; using it now";
		Redo_log.switch ha_redo_log Xapi_globs.gen_metadata_vdi_reason
	end
	else begin
		Redo_log_usage.stop_using_redo_log ha_redo_log;
		Redo_log.disable ha_redo_log
	end

(* This function is called when HA is found to be enabled at startup, before
 * the DB backend is initialised. Read the latest DB from the block-device, and
 * make future DB changes get written as deltas. *)
let redo_log_ha_enabled_at_startup () =
	(* If we are still the master, extract any HA metadata database so we can consider population from it *)
	if Pool_role.is_master () then begin
		debug "HA is enabled, so enabling writing to redo-log";
		Redo_log.enable ha_redo_log Xapi_globs.ha_metadata_vdi_reason; (* enable the use of the redo log *)
		debug "This node is a master; attempting to extract a database from a metadata VDI";
		let db_ref = Db_backend.make () in 
		Redo_log_usage.read_from_redo_log ha_redo_log Xapi_globs.ha_metadata_db db_ref (* best effort only: does not raise any exceptions *)
	end

(* ----------------------------- *)


(** Called when xapi restarts: server may be in emergency mode at this point. We need
	to inspect the local configuration and if HA is supposed to be armed we need to
	set everything up.
	Note that
	the master shouldn't be able to activate HA while we are offline since that would cause
	us to come up with a broken configuration (the enable-HA stage has the critical task of
	synchronising the HA configuration on all the hosts). So really we only want to notice
	if the Pool has had HA disabled while we were offline. *)
let on_server_restart () =
	let armed = bool_of_string (Localdb.get Constants.ha_armed) in

	if armed then begin
		debug "HA is supposed to be armed";
		(* Make sure daemons are up *)

		let finished = ref false in
		(* Do not proceed any further until the situation is resolved.
		   XXX we might need some kind of user-override *)
		while not (!finished) do

			(* If someone has called Host.emergency_ha_disable in the background then we notice the change here *)
			if not (try bool_of_string (Localdb.get Constants.ha_armed) with _ -> false) then begin
				warn "ha_start_daemon aborted because someone has called Host.emergency_ha_disable";
				failwith "Host.emergency_ha_disable"; (* failure causes HA startup to abort *)
			end;

			try
				if Xapi_fist.ha_cannot_access_statefile () then raise (Xha_error Xha_errno.Mtc_exit_can_not_access_statefile);
				if Xapi_fist.ha_daemon_startup_failed () then failwith "simulating xha daemon startup failure";

				(* CA-21406: Try again to reattach the statefile VDI *)
				Static_vdis.reattempt_on_boot_attach ();

				let (_ : string) = call_script ha_start_daemon [] in
				finished := true;
			with
				| Xha_error Xha_errno.Mtc_exit_daemon_is_present ->
					warn "ha_start_daemon failed with MTC_EXIT_DAEMON_IS_PRESENT: continuing with startup";
					finished := true;
				| Xha_error Xha_errno.Mtc_exit_invalid_pool_state as e ->
					warn "ha_start_daemon failed with MTC_EXIT_INVALID_POOL_STATE: disabling HA on this host";
					Localdb.put Constants.ha_armed "false";
					raise e
				| Xha_error Xha_errno.Mtc_exit_can_not_access_statefile as e ->
					warn "ha_start_daemon failed with MTC_EXIT_CAN_NOT_ACCESS_STATEFILE: will contact existing master and check if HA is still enabled";

					(* check the Pool.ha_enabled on the master... assuming we can find the master. If we can't we stay here forever. *)
					let master_can_confirm_ha_is_disabled () =
						try
							let address = Pool_role.get_master_address () in
							Helpers.call_emergency_mode_functions address
								(fun rpc session_id ->
									let pool = List.hd (Client.Pool.get_all rpc session_id) in
									not (Client.Pool.get_ha_enabled rpc session_id pool)
								)
						with _ ->
							(* there's no-one for us to ask about whether HA is enabled or not *)
							false in
					if master_can_confirm_ha_is_disabled () then begin
						info "Existing master confirmed that HA is disabled pool-wide: disabling HA on this host";
						Localdb.put Constants.ha_armed "false";
						raise e
					end;
					info "Assuming HA is still enabled on the Pool and that our storage system has failed: will retry in 10s";
					Xapi_globs.slave_emergency_mode := true;
					Xapi_globs.emergency_mode_error := Api_errors.Server_error(Api_errors.ha_host_cannot_access_statefile, []);
					Helpers.touch_file !Xapi_globs.ready_file;
					Thread.delay 10.;
				| Xha_error errno ->
					error "ha_start_daemon failed with unexpected error %s: will retry in 10s" (Xha_errno.to_string errno);
					Xapi_globs.slave_emergency_mode := true;
					Xapi_globs.emergency_mode_error := Api_errors.Server_error(Api_errors.ha_heartbeat_daemon_startup_failed, []);
					Helpers.touch_file !Xapi_globs.ready_file;
					Thread.delay 10.;
				| e ->
					error "ha_start_daemon failed with unexpected exception: %s -- retrying in 10s" (ExnHelper.string_of_exn e);
					Xapi_globs.slave_emergency_mode := true;
					Xapi_globs.emergency_mode_error := Api_errors.Server_error(Api_errors.ha_heartbeat_daemon_startup_failed, []);
					Helpers.touch_file !Xapi_globs.ready_file;
					Thread.delay 10.;
		done;

		if Pool_role.is_master () then begin
			if not (local_failover_decisions_are_ok ()) then begin
				warn "ha.disable_failover_decisions flag set: not proposing myself as master";
				on_master_failure ()
			end else begin
				if propose_master ()
				then info "ha_propose_master succeeded; continuing"
				else on_master_failure ()
			end
		end;

		(* Start up the redo-log if appropriate. *)
		redo_log_ha_enabled_at_startup ();

		debug "About to start the monitor";
		Monitor.start ();
		(* We signal the monitor that the database state is valid (wrt liveness + disabledness of hosts) later *)
	end

(** Called in the master xapi startup when the database is ready. We set all hosts (including this one) to
	disabled then signal the monitor thread to look. It can then wait for slaves to turn up
	before trying to restart VMs. *)
let on_database_engine_ready () =
	info "Setting all hosts to dead and disabled. Hosts must re-enable themselves explicitly";
	Server_helpers.exec_with_new_task "Setting all hosts to dead and disabled"
		(fun __context ->
			List.iter (fun self ->
				let uuid = Db.Host.get_uuid ~__context ~self in
				let hostname = Db.Host.get_hostname ~__context ~self in
				info "Host.enabled: on_database_engine_ready: setting host %s (%s) to disabled" uuid hostname;
				Db.Host.set_enabled ~__context ~self ~value:false
			)
				(Db.Host.get_all ~__context)
		);
	Monitor.signal_database_state_valid ()

(*********************************************************************************************)
(* Internal API calls to configure individual hosts                                          *)

(** Internal API call to prevent this node making an unsafe failover decision.
	This call is idempotent. *)
let ha_disable_failover_decisions __context localhost =
	debug "Disabling failover decisions";
	(* FIST *)
	if Xapi_fist.disable_ha_disable_failover () then begin
		error "FIST: ha_disable_failover_decisions";
		failwith "FIST: ha_disable_failover_decisions"
	end;
	Localdb.put Constants.ha_disable_failover_decisions "true"

(** Internal API call to disarm localhost.
	If the daemon is missing then we return success. Either fencing was previously disabled and the
	daemon has shutdown OR the daemon has died and this node will fence shortly...
 *)
let ha_disarm_fencing __context localhost =
	try
		let (_ : string) = call_script ha_disarm_fencing [] in ()
	with Xha_error Xha_errno.Mtc_exit_daemon_is_not_present ->
		info "ha_disarm_fencing: daemon has exited so returning success"

let ha_set_excluded __context localhost =
	let (_ : string) = call_script ha_set_excluded [] in ()

(** Internal API call to stop the HA daemon.
	This call is idempotent. *)
let ha_stop_daemon __context localhost =
	Monitor.stop ();
	let (_ : string) = call_script ha_stop_daemon [] in ()

(** Emergency-mode API call to disarm localhost *)
let emergency_ha_disable __context =
	if Localdb.get Constants.ha_armed = "false"
	then raise (Api_errors.Server_error(Api_errors.ha_not_enabled, []));

	warn "Host.emergency_ha_disable: Disabling the HA subsystem on the local host only.";
	Localdb.put Constants.ha_armed "false";

	begin
		try
			ha_disarm_fencing __context ();
		with Xha_error e ->
			error "Host.emergency_ha_disable: ha_disarm_fencing failed with %s; continuing" (Xha_errno.to_string e)
	end;
	begin
		try
			ha_stop_daemon __context ();
		with Xha_error e ->
			error "Host.emergency_ha_disable: ha_stop_daemon failed with %s; continuing" (Xha_errno.to_string e)
	end;
	(* Might not be able to access the database to detach statefiles; however this isn't critical *)
	()

(** Internal API call to release any HA resources after the system has
	been shutdown.  This call is idempotent. Modified for CA-48539 to
	call vdi.deactivate before vdi.detach. *)
let ha_release_resources __context localhost =
	Monitor.stop ();

	(* Why aren't we calling Xha_statefile.detach_existing_statefiles?
	   Does Db.Pool.get_ha_statefiles return a different set of
	   statefiles than Xha_statefile.list_existing_statefiles? *)

	(* Deactivate and detach all statefile VDIs in the entire pool *)
	let statefile_vdis = Db.Pool.get_ha_statefiles ~__context ~self:(Helpers.get_pool ~__context)
	and deactiavte_and_detach_vdi vdi_str =
		let uuid = Db.VDI.get_uuid ~__context ~self:(Ref.of_string vdi_str) in
		Helpers.log_exn_continue
			(Printf.sprintf "detaching statefile VDI uuid: %s" uuid)
			(fun () ->
				Static_vdis.permanent_vdi_deactivate_by_uuid ~__context ~uuid ;
				Static_vdis.permanent_vdi_detach_by_uuid ~__context ~uuid) ()
	in List.iter deactiavte_and_detach_vdi statefile_vdis ;

	(* Deactivate and detach any metadata VDIs *)
	Helpers.log_exn_continue
		(Printf.sprintf "deactivating and detaching metadata VDIs")
		(fun () -> Xha_metadata_vdi.deactivate_and_detach_existing ~__context) ();

	(* At this point a restart won't enable the HA subsystem *)
	Localdb.put Constants.ha_armed "false"

(** Internal API call which blocks until this node's xHA daemon spots the invalid statefile
	and exits cleanly. If the daemon survives but the statefile access is lost then this function
	will return an exception and the no-statefile shutdown can be attempted.
 *)
let ha_wait_for_shutdown_via_statefile __context localhost =
	try
		while true do
			let liveset = query_liveset () in

			let hosts = liveset.Xha_interface.LiveSetInformation.hosts in
			let me = Hashtbl.find hosts liveset.Xha_interface.LiveSetInformation.local_host_id in
			(* If we have no statefile access or if it looks corrupted, fail the operation *)
			if false
				|| not me.Xha_interface.LiveSetInformation.Host.state_file_access
				|| me.Xha_interface.LiveSetInformation.Host.state_file_corrupted
			then raise (Api_errors.Server_error(Api_errors.ha_lost_statefile, []));

			(* check to see if this node still has statefile access *)
			Thread.delay 5.
		done
	with
		| Xha_error Xha_errno.Mtc_exit_daemon_is_not_present ->
			info "ha_wait_for_shutdown_via_statefile: daemon has exited so returning success"


(** Attach the statefile VDIs and return the resulting list of paths in dom0 *)
let attach_statefiles ~__context statevdis =
	(* First GC any existing statefiles: these are not needed any more *)
	info "Detaching any existing statefiles: these are not needed any more";
	Xha_statefile.detach_existing_statefiles ~__context;

	let paths = ref [] in
	begin
		let cur_vdi_str = ref "" in
		try
			List.iter
				(fun vdi ->
					cur_vdi_str := Ref.string_of vdi;
					info "Attempting to permanently attach statefile VDI: %s" (Ref.string_of vdi);
					paths := Static_vdis.permanent_vdi_attach ~__context ~vdi ~reason:Xha_statefile.reason:: !paths) statevdis
		with e ->
			error "Caught exception attaching statefile: %s" (ExnHelper.string_of_exn e);
			List.iter
				(fun vdi ->
					Helpers.log_exn_continue
						(Printf.sprintf "detaching statefile: %s" (Ref.string_of vdi))
						(fun () -> Static_vdis.permanent_vdi_detach ~__context ~vdi) ()
				) statevdis;
			raise  (Api_errors.Server_error(Api_errors.vdi_not_available, [!cur_vdi_str]))
	end;
	!paths

(** Attach the metadata VDI and return the resulting path in dom0 *)
let attach_metadata_vdi ~__context vdi =
	info "Detaching any existing metadata volume: these are not needed anymore";
	Xha_metadata_vdi.detach_existing ~__context;
{'fraction_non_alphanumeric': 0.07898236358483574, 'fraction_numerical': 0.003215539229578601, 'mean_word_length': 3.758454356846473, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 211, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '315073', 'n_tokens_mistral': 32959, 'n_tokens_neox': 27937, 'n_words': 9318}