Merge pull request #59 from jonludlam/CP-12141-finally

CP-12141: Remove master/slave knowledge from RRDD

Author: Jon Ludlam

lib/pool_role_shared.ml

@@ -2,5 +2,4 @@
 # DO NOT EDIT (digest: 0c7354c835dd0cb4479d655ef155db47)
 Stats
 Constants
-Pool_role_shared
 # OASIS_STOP

lib/rrdd_libs.mldylib

@@ -2,5 +2,4 @@
 # DO NOT EDIT (digest: 0c7354c835dd0cb4479d655ef155db47)
 Stats
 Constants
-Pool_role_shared
 # OASIS_STOP

lib/rrdd_libs.mllib

@@ -125,7 +125,7 @@ let put_rrd_handler (req : Http.Request.t) (s : Unix.file_descr) _ =
 	if List.mem_assoc "archive" query then (
 		debug "Receiving RRD on the master for archiving, type=%s."
 			(if is_host then "Host" else "VM uuid=" ^ uuid);
-		archive_rrd ~uuid ~rrd:(Rrd.copy_rrd rrd) ()
+		archive_rrd_internal ~uuid ~rrd:(Rrd.copy_rrd rrd) ()
 	) else (
 		debug "Receiving RRD for resident VM uuid=%s. Replacing in hashtable." uuid;
 		let domid = int_of_string (List.assoc "domid" query) in

rrdd/rrdd_http_handler.ml

@@ -39,28 +39,18 @@ let create_fresh_rrd use_min_max dss =
  * domain has gone and we stream the RRD to the master. We also have a
  * list of the currently rebooting VMs to ensure we don't accidentally
  * archive the RRD. *)
-let update_rrds timestamp dss (uuid_domids : (string * int) list) rebooting_vms paused_vms =
+let update_rrds timestamp dss (uuid_domids : (string * int) list) paused_vms =
 	(* Here we do the synchronising between the dom0 view of the world
 		 and our Hashtbl. By the end of this execute block, the Hashtbl
 		 correctly represents the world *)
-	let to_send_back = Mutex.execute mutex (fun _ ->
+	Mutex.execute mutex (fun _ ->
 		let out_of_date, by_how_much =
 			match !host_rrd with
 			| None -> false, 0.
 			| Some rrdi -> rrdi.rrd.Rrd.last_updated > timestamp, abs_float (timestamp -. rrdi.rrd.Rrd.last_updated)
 		in
 		if out_of_date then
 			error "Clock just went backwards by %.0f seconds: RRD data may now be unreliable" by_how_much;
-		let registered = Hashtblext.fold_keys vm_rrds in
-		let gone_vms = List.filter (fun vm -> not (List.mem_assoc vm uuid_domids)) registered in
-		let to_send_back = List.map (fun uuid -> uuid, Hashtbl.find vm_rrds uuid) gone_vms in
-		(* Don't send back rebooting VMs! *)
-		let to_send_back = List.filter (fun (uuid, _) ->
-			let rebooting = (List.exists (fun uuid' -> uuid = uuid') rebooting_vms) in
-			if rebooting then debug "Ignoring disappeared VM which is rebooting";
-			not rebooting
-		) to_send_back in
-		List.iter (fun (uuid, _) -> Hashtbl.remove vm_rrds uuid) to_send_back;
 		let do_vm (vm_uuid, domid) =
 			try
 				let dss = List.filter_map (fun (ty, ds) -> match ty with | VM x -> if x = vm_uuid then Some ds else None | _ -> None) dss in
@@ -169,8 +159,4 @@ let update_rrds timestamp dss (uuid_domids : (string * int) list) rebooting_vms
 				Rrd.ds_update_named rrd timestamp ~new_domid:false
 					(List.map (fun ds -> (ds.ds_name, (ds.ds_value,ds.ds_pdp_transform_function))) host_dss)
 		end;
-		to_send_back
 	)
-	in List.iter (fun (uuid, rrdi) ->
-		debug "Sending back RRD for VM uuid=%s" uuid;
-		archive_rrd ~uuid ~rrd:rrdi.rrd ()) to_send_back

rrdd/rrdd_monitor.ml

@@ -34,7 +34,7 @@ let archive_sr_rrd _ ~(sr_uuid : string) : string =
 			raise (Archive_failed(msg))
 	) in
 	try
-		archive_rrd ~uuid:sr_uuid ~rrd:sr_rrd.rrd ();
+		archive_rrd_internal ~uuid:sr_uuid ~rrd:sr_rrd.rrd ();
 		let archive_path =
 			Filename.concat Constants.rrd_location (sr_uuid ^ ".gz") in
 		if not (Unixext.file_exists archive_path) then begin
@@ -62,8 +62,18 @@ let push_sr_rrd _ ~(sr_uuid : string) ~(path : string) : unit =
 let has_vm_rrd _ ~(vm_uuid : string) =
 	Mutex.execute mutex (fun _ -> Hashtbl.mem vm_rrds vm_uuid)
 
-let backup_rrds _ ?(save_stats_locally = true) () : unit =
-	debug "backup safe_stats_locally=%b" save_stats_locally;
+let archive_rrd _ ~vm_uuid ~remote_address : unit =
+	Mutex.execute mutex (fun () ->
+		try
+			let rrd = (Hashtbl.find vm_rrds vm_uuid).rrd in
+			archive_rrd_internal ~remote_address ~uuid:vm_uuid ~rrd ();
+			Hashtbl.remove vm_rrds vm_uuid
+		with Not_found -> ())
+
+let backup_rrds _ ?(remote_address = None) () : unit =
+	debug "backing up rrds %s" (match remote_address with
+		| None -> "locally"
+		| Some x -> Printf.sprintf "remotely at %s" x);
 	let total_cycles = 5 in
 	let cycles_tried = ref 0 in
 	while !cycles_tried < total_cycles do
@@ -81,7 +91,7 @@ let backup_rrds _ ?(save_stats_locally = true) () : unit =
 				(fun (uuid, rrd) ->
 					debug "Backup: saving RRD for VM uuid=%s to local disk" uuid;
 					let rrd = Mutex.execute mutex (fun () -> Rrd.copy_rrd rrd) in
-					archive_rrd ~save_stats_locally ~uuid ~rrd ()
+					archive_rrd_internal ~remote_address ~uuid ~rrd ()
 				) vrrds;
 			let srrds =
 				try
@@ -95,13 +105,13 @@ let backup_rrds _ ?(save_stats_locally = true) () : unit =
 				(fun (uuid, rrd) ->
 					debug "Backup: saving RRD for SR uuid=%s to local disk" uuid;
 					let rrd = Mutex.execute mutex (fun () -> Rrd.copy_rrd rrd) in
-					archive_rrd ~uuid ~rrd ()
+					archive_rrd_internal ~uuid ~rrd ()
 				) srrds;
 			match !host_rrd with
 			| Some rrdi ->
 				debug "Backup: saving RRD for host to local disk";
 				let rrd = Mutex.execute mutex (fun () -> Rrd.copy_rrd rrdi.rrd) in
-				archive_rrd ~save_stats_locally ~uuid:(Inventory.lookup Inventory._installation_uuid) ~rrd ()
+				archive_rrd_internal ~remote_address ~uuid:(Inventory.lookup Inventory._installation_uuid) ~rrd ()
 			| None -> ()
 		end else begin
 			cycles_tried := 1 + !cycles_tried;
@@ -120,17 +130,16 @@ let load_rrd_from_local_filesystem uuid =
 module Deprecated = struct
 	(* DEPRECATED *)
 	(* Fetch an RRD from the master *)
-	let pull_rrd_from_master ~uuid ~is_host =
+	let pull_rrd_from_master ~uuid ~master_address =
 		let pool_secret = get_pool_secret () in
-		let uri = if is_host then Constants.get_host_rrd_uri else Constants.get_vm_rrd_uri in
+		let uri = Constants.get_host_rrd_uri in
 		(* Add in "dbsync = true" to the query to make sure the master
 		 * doesn't try to redirect here! *)
 		let uri = uri ^ "?uuid=" ^ uuid ^ "&dbsync=true" in
 		let request =
 			Http.Request.make ~user_agent:Constants.rrdd_user_agent
 			~cookie:["pool_secret", pool_secret] Http.Get uri in
 		let open Xmlrpc_client in
-		let master_address = Pool_role_shared.get_master_address () in
 		let transport = SSL(SSL.make (), master_address, !Rrdd_shared.https_port) in
 		with_transport transport (
 			with_http request (fun (response, s) ->
@@ -149,60 +158,56 @@ module Deprecated = struct
 	 * 1. For the local host after a xapi restart or host restart.
 	 * 2. For running VMs after a xapi restart.
 	 * It is now only used to load the host's RRD after xapi restart. *)
-	let load_rrd _ ~(uuid : string) ~(domid : int) ~(is_host : bool)
-			~(timescale : int) () : unit =
+	let load_rrd _ ~(uuid : string) ~(timescale : int) ~(master_address : string option) : unit =
 		try
 			let rrd =
 				try
 					let rrd = load_rrd_from_local_filesystem uuid in
 					debug "RRD loaded from local filesystem for object uuid=%s" uuid;
 					rrd
 				with e ->
-					if Pool_role_shared.is_master () then begin
+					match master_address with
+					| None -> begin
 						info "Failed to load RRD from local filesystem: metrics not available for uuid=%s" uuid;
 						raise e
-					end else begin
+					end
+					| Some x -> begin
 						debug "Failed to load RRD from local filesystem for object uuid=%s; asking master" uuid;
 						try
-							let rrd = pull_rrd_from_master ~uuid ~is_host in
+							let rrd = pull_rrd_from_master ~uuid ~master_address:x in
 							debug "RRD pulled from master for object uuid=%s" uuid;
 							rrd
 						with e ->
 							info "Failed to fetch RRD from master: metrics not available for uuid=%s" uuid;
 							raise e
 					end
 			in
-			Mutex.execute mutex (fun () ->
-				if is_host
-				then begin
-					host_rrd := Some {rrd; dss = []; domid}
-				end else
-					Hashtbl.replace vm_rrds uuid {rrd; dss = []; domid}
-			)
+			Mutex.execute mutex (fun () -> host_rrd := Some {rrd; dss = []; domid = 0} )
 		with _ -> ()
 end
 
-(* Push function to push the archived RRD to the appropriate host
- * (which might be us, in which case, pop it into the hashtbl. *)
-let push_rrd _ ~(vm_uuid : string) ~(domid : int) ~(is_on_localhost : bool) ()
-		: unit =
+let get_rrd ~vm_uuid =
+	let path = Filename.concat Constants.rrd_location vm_uuid in
+	rrd_of_gzip path
+
+let push_rrd_local _ ~vm_uuid ~domid : unit =
 	try
-		let path = Constants.rrd_location ^ "/" ^ vm_uuid in
-		let rrd = rrd_of_gzip path in
-		debug "Pushing RRD for VM uuid=%s" vm_uuid;
-		if is_on_localhost then
-			Mutex.execute mutex (fun _ ->
-				Hashtbl.replace vm_rrds vm_uuid {rrd; dss=[]; domid}
-			)
-		else
-			(* Host might be OpaqueRef:null, in which case we'll fail silently *)
-			let address = Pool_role_shared.get_master_address () in
-			send_rrd ~address ~to_archive:false ~uuid:vm_uuid
-				~rrd:(Rrd.copy_rrd rrd) ()
+		let rrd = get_rrd ~vm_uuid in
+		debug "Pushing RRD for VM uuid=%s locally" vm_uuid;
+		Mutex.execute mutex (fun _ ->
+			Hashtbl.replace vm_rrds vm_uuid {rrd; dss=[]; domid}
+		)
+	with _ -> ()
+
+let push_rrd_remote _ ~vm_uuid ~remote_address : unit =
+	try
+		let rrd = get_rrd ~vm_uuid in
+		debug "Pushing RRD for VM uuid=%s remotely" vm_uuid;
+		send_rrd ~address:remote_address ~to_archive:false ~uuid:vm_uuid ~rrd:(Rrd.copy_rrd rrd) ()
 	with _ -> ()
 
 (** Remove an RRD from the local filesystem, if it exists. *)
-let remove_rrd _ ~(uuid : string) () : unit =
+let remove_rrd _ ~(uuid : string) : unit =
 	let path = Constants.rrd_location ^ "/" ^ uuid in
 	let gz_path = path ^ ".gz" in
 	(try Unix.unlink path with _ -> ());
@@ -215,7 +220,7 @@ let remove_rrd _ ~(uuid : string) () : unit =
  * Remote address is assumed to be valid, since it is set by monitor_master.
  *)
 let migrate_rrd _ ?(session_id : string option) ~(remote_address : string)
-		~(vm_uuid : string) ~(host_uuid : string) () : unit =
+		~(vm_uuid : string) ~(host_uuid : string) : unit =
 	try
 		let rrdi = Mutex.execute mutex (fun () ->
 			let rrdi = Hashtbl.find vm_rrds vm_uuid in
@@ -237,12 +242,12 @@ let migrate_rrd _ ?(session_id : string option) ~(remote_address : string)
 
 (* Called on host shutdown/reboot to send the Host RRD to the master for
  * backup. Note all VMs will have been shutdown by now. *)
-let send_host_rrd_to_master _ () =
+let send_host_rrd_to_master _ ~master_address =
 	match !host_rrd with
 	| Some rrdi ->
 		debug "sending host RRD to master";
 		let rrd = Mutex.execute mutex (fun () -> Rrd.copy_rrd rrdi.rrd) in
-		archive_rrd ~save_stats_locally:false ~uuid:(Inventory.lookup Inventory._installation_uuid) ~rrd ()
+		send_rrd ~address:master_address ~to_archive:true ~uuid:(Inventory.lookup Inventory._installation_uuid) ~rrd ()
 	| None -> ()
 
 let add_ds ~rrdi ~ds_name =
@@ -346,7 +351,7 @@ let query_sr_ds _ ~(sr_uuid : string) ~(ds_name : string) : float =
 		Rrd.query_named_ds rrdi.rrd now ds_name Rrd.CF_Average
 	)
 
-let update_use_min_max _ ~(value : bool) () : unit =
+let update_use_min_max _ ~(value : bool) : unit =
 	debug "Updating use_min_max: New value=%b" value;
 	use_min_max := value
 
@@ -357,7 +362,7 @@ let update_vm_memory_target _ ~(domid : int) ~(target : int64) : unit =
 	Mutex.execute memory_targets_m
 		(fun _ -> Hashtbl.replace memory_targets domid target)
 
-let set_cache_sr _ ~(sr_uuid : string) () : unit =
+let set_cache_sr _ ~(sr_uuid : string) : unit =
 	Mutex.execute cache_sr_lock (fun () -> cache_sr_uuid := Some sr_uuid)
 
 let unset_cache_sr _ () =
@@ -529,7 +534,7 @@ end
 
 module HA = struct
 	let enable_and_update _ ~(statefile_latencies : Rrd.Statefile_latency.t list)
-			~(heartbeat_latency : float) ~(xapi_latency : float) () =
+			~(heartbeat_latency : float) ~(xapi_latency : float) =
 		Mutex.execute Rrdd_ha_stats.m (fun _ ->
 			Rrdd_ha_stats.enabled := true;
 			Rrdd_ha_stats.Statefile_latency.all := statefile_latencies;

rrdd/rrdd_server.ml

@@ -120,11 +120,10 @@ let send_rrd ?(session_id : string option) ~(address : string)
 	);
 	debug "Sending RRD complete."
 
-let archive_rrd ?(save_stats_locally = Pool_role_shared.is_master ()) ~uuid
-		~rrd () =
-	debug "Archiving RRD for object uuid=%s %s" uuid
-		(if save_stats_locally then "to local disk" else "to remote master");
-	if save_stats_locally then begin
+let archive_rrd_internal ?(remote_address = None) ~uuid ~rrd () =
+	match remote_address with
+	| None -> begin
+		debug "Archiving RRD for object uuid=%s to local disk" uuid;
 		try
 			(* Stash away the rrd onto disk. *)
 			let exists =
@@ -146,9 +145,9 @@ let archive_rrd ?(save_stats_locally = Pool_role_shared.is_master ()) ~uuid
 		with e ->
 			(*debug "Caught exception: %s" (ExnHelper.string_of_exn e);*)
 			log_backtrace();
-	end else begin
+	end
+	| Some x -> begin
 		(* Stream it to the master to store, or maybe to a host in the migrate case *)
-		debug "About to send to master.";
-		let address = Pool_role_shared.get_master_address () in
-		send_rrd ~address ~to_archive:true ~uuid ~rrd ()
+		debug "Archiving RRD for object uuid=%s to remote master" uuid;
+		send_rrd ~address:x ~to_archive:true ~uuid ~rrd ()
 	end