<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.
 *)

open Stdext
open Client
open Db_filter_types
open Pervasiveext
open Threadext
open Xstringext
open Listext
open Network

module L = Debug.Make(struct let name="license" end)
module D=Debug.Make(struct let name="xapi" end)
open D
open Workload_balancing

(* Surpress exceptions *)
let no_exn f x =
  try ignore (f x)
  with exn ->
    debug "Ignoring exception: %s" (ExnHelper.string_of_exn exn)

let rpc host_address xml =
  try
    Helpers.make_remote_rpc host_address xml
  with Xmlrpc_client.Connection_reset ->
    raise (Api_errors.Server_error(Api_errors.pool_joining_host_connection_failed, []))

let get_master ~rpc ~session_id =
  let pool = List.hd (Client.Pool.get_all rpc session_id) in
  Client.Pool.get_master rpc session_id pool

(* Pre-join asserts *)
let pre_join_checks ~__context ~rpc ~session_id ~force =
  (* I cannot join a Pool unless my management interface exists in the db, otherwise
     	   Pool.eject will fail to rewrite network interface files. *)
  let assert_management_interface_exists () =
    try
      let (_: API.ref_PIF) = Xapi_host.get_management_interface ~__context ~host:(Helpers.get_localhost ~__context) in
      ()
    with _ ->
      error "Pool.join/Pool.eject requires a properly configured management interface. Wait for xapi/firstboot initialisation to complete and then retry.";
      raise (Api_errors.Server_error(Api_errors.host_still_booting, [])) in

  (* I cannot join a Pool if I have HA already enabled on me *)
  let ha_is_not_enable_on_me () =
    let pool = Helpers.get_pool ~__context in
    if Db.Pool.get_ha_enabled ~__context ~self:pool then begin
      error "Cannot join pool as HA is enabled";
      raise (Api_errors.Server_error(Api_errors.ha_is_enabled, []))
    end in

  (* I Cannot join a Pool if it has HA enabled on it *)
  let ha_is_not_enable_on_the_distant_pool () =
    let pool = List.hd (Client.Pool.get_all rpc session_id) in
    if Client.Pool.get_ha_enabled rpc session_id pool then begin
      error "Cannot join pool which already has HA enabled";
      raise (Api_errors.Server_error(Api_errors.ha_is_enabled, []));
    end in

  (* CA-26975: Pool edition MUST match *)
  let assert_restrictions_match () =
    let my_edition = Db.Host.get_edition ~__context ~self:(Helpers.get_localhost ~__context) in
    let host_records = List.map snd (Client.Host.get_all_records ~rpc ~session_id) in
    let pool_editions = List.map (fun host_r -> host_r.API.host_edition) host_records in
    (* If all hosts have the same edition string, we need do no more. *)
    if List.fold_left (fun b edn -> edn = my_edition && b) true pool_editions then ()
    else try
        (* We have different edition strings so must consult v6d for their significance.
           			 * This will fail with v6d_failure if v6d is not running. *)
        let editions = V6_client.get_editions "assert_restrictions_match" in
        let edition_to_int e =
          try
            match List.find (fun (name, _) -> name = e) editions with _, (_, _, a) -> a
          with Not_found ->
            (* Happens if pool has edition "free/libre" (no v6d) *)
            error "Pool.join failed: pool has a host with edition unknown to v6d: %s" e;
            raise (Api_errors.Server_error(Api_errors.license_host_pool_mismatch,
                                           ["Edition \"" ^ e ^ "\" from pool is not known to v6d."]))
        in
        let min_edition l =
          List.fold_left (fun m e -> if edition_to_int e < edition_to_int m then e else m) (List.hd l) l
        in
        (* get pool edition: the "minimum" edition among all hosts *)
        let pool_edition = min_edition pool_editions in
        (* compare my edition to pool edition *)
        if (edition_to_int pool_edition) <> (edition_to_int my_edition) then begin
          error "Pool.join failed due to edition mismatch";
          error "Remote has %s" pool_edition;
          error "Local has  %s" my_edition;
          raise (Api_errors.Server_error(Api_errors.license_host_pool_mismatch,
                                         ["host edition = \""^my_edition^"\""; "pool edition = \""^pool_edition^"\""]))
        end
      with Api_errors.Server_error (code, []) when code = Api_errors.v6d_failure ->
        error "Pool.join failed because edition strings differ and local has no license daemon running.";
        let pool_edn_list_str = "[" ^ (String.concat "; " pool_editions) ^ "]" in
        error "Remote editions: %s" pool_edn_list_str;
        error "Local edition: %s" my_edition;
        raise (Api_errors.Server_error (code, ["The pool uses v6d. Pool edition list = " ^ pool_edn_list_str]))
  in

  let assert_api_version_matches () =
    let master = get_master rpc session_id in
    let candidate_slave = Helpers.get_localhost ~__context in
    let master_major = Client.Host.get_API_version_major ~rpc ~session_id ~self:master in
    let master_minor = Client.Host.get_API_version_minor ~rpc ~session_id ~self:master in
    let slave_major = Db.Host.get_API_version_major ~__context ~self:candidate_slave in
    let slave_minor = Db.Host.get_API_version_minor ~__context ~self:candidate_slave in
    if master_major <> slave_major || master_minor <> slave_minor then
    begin
      error "The joining host's API version is %Ld.%Ld while the master's is %Ld.%Ld"
        slave_major slave_minor master_major master_minor;
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_must_have_same_api_version,
        [Printf.sprintf "%Ld.%Ld" slave_major slave_minor; Printf.sprintf "%Ld.%Ld" master_major master_minor;]))
    end
  in

  let assert_db_schema_matches () =
    let master = get_master rpc session_id in
    let candidate_slave = Helpers.get_localhost ~__context in
    let master_sw_version = Client.Host.get_software_version ~rpc ~session_id ~self:master in
    let slave_sw_version = Db.Host.get_software_version ~__context ~self:candidate_slave in
    let master_db_schema = try List.assoc Xapi_globs._db_schema master_sw_version with _ -> "" in
    let slave_db_schema = try List.assoc Xapi_globs._db_schema slave_sw_version with _ -> "" in
    if master_db_schema = "" || slave_db_schema = "" ||  master_db_schema <> slave_db_schema then
    begin
        error "The joining host's database schema is %s; the master's is %s"
          slave_db_schema master_db_schema;
        raise (Api_errors.Server_error(Api_errors.pool_joining_host_must_have_same_db_schema,
          [slave_db_schema; master_db_schema]))
    end
  in

  (* CP-700: Restrict pool.join if AD configuration of slave-to-be does not match *)
  (* that of master of pool-to-join *)
  let assert_external_auth_matches () =
    let master = get_master rpc session_id in
    let slavetobe = Helpers.get_localhost ~__context in
    let slavetobe_auth_type = Db.Host.get_external_auth_type ~__context ~self:slavetobe in
    let slavetobe_auth_service_name = Db.Host.get_external_auth_service_name ~__context ~self:slavetobe in
    let master_auth_type = Client.Host.get_external_auth_type ~rpc ~session_id ~self:master in
    let master_auth_service_name = Client.Host.get_external_auth_service_name ~rpc ~session_id ~self:master in
    debug "Verifying if external auth configuration of master %s (auth_type=%s service_name=%s) matches that of slave-to-be %s (auth-type=%s service_name=%s)"
      (Client.Host.get_name_label ~rpc ~session_id ~self:master) master_auth_type master_auth_service_name
      (Db.Host.get_name_label ~__context ~self:slavetobe) slavetobe_auth_type slavetobe_auth_service_name;
    if (slavetobe_auth_type <> master_auth_type)
    || (slavetobe_auth_service_name <> master_auth_service_name) then begin
      error "Cannot join pool whose external authentication configuration is different";
      raise (Api_errors.Server_error(Api_errors.pool_joining_external_auth_mismatch, []))
    end in

  let assert_i_know_of_no_other_hosts () =
    let hosts = Db.Host.get_all ~__context in
    if List.length hosts > 1 then begin
      error "The current host is already the master of other hosts: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_be_master_of_other_hosts, []))
    end in

  let assert_no_running_vms_on_me () =
    let my_vms = Db.VM.get_all_records ~__context in
    let my_running_vms =
      List.filter
        (fun (_,vmrec) -> (not (Helpers.is_domain_zero ~__context (Db.VM.get_by_uuid ~__context ~uuid:vmrec.API.vM_uuid)))
                          && vmrec.API.vM_power_state = `Running) my_vms in
    if List.length my_running_vms > 0 then begin
      error "The current host has running or suspended VMs: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_have_running_VMs, []))
    end in

  let assert_no_vms_with_current_ops () =
    let my_vms = Db.VM.get_all_records ~__context in
    let vms_with_current_ops =
      List.filter (fun (_,vmr) -> (List.length vmr.API.vM_current_operations)>0 ) my_vms in
    if List.length vms_with_current_ops > 0 then begin
      error "The current host has VMs with current operations: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_have_vms_with_current_operations, []))
    end in

  let assert_no_shared_srs_on_me () =
    let my_srs = Db.SR.get_all_records ~__context in
    let my_shared_srs = List.filter (fun (sr,srec)-> srec.API.sR_shared && not srec.API.sR_is_tools_sr) my_srs in
    if not (my_shared_srs = []) then begin
      error "The current host has shared SRs: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_contain_shared_SRs, []))
    end in

  (* Allow pool-join if host does not have any bonds *)
  let assert_no_bonds_on_me () =
    if Db.Bond.get_all ~__context <> [] then begin
      error "The current host has network bonds: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_has_bonds, []))
    end in

  (* Allow pool-join if host does not have any tunnels *)
  let assert_no_tunnels_on_me () =
    if Db.Tunnel.get_all ~__context <> [] then begin
      error "The current host has tunnels: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_has_tunnels, []))
    end in

  (* Allow pool-join if host does not have any non-management VLANs *)
  let assert_no_non_management_vlans_on_me () =
    List.iter (fun self ->
      let pif = Db.VLAN.get_untagged_PIF ~__context ~self in
      if Db.PIF.get_management ~__context ~self:pif <> true then begin
        error "The current host has non-management vlans: it cannot join a new pool";
        raise (Api_errors.Server_error(Api_errors.pool_joining_host_has_non_management_vlans, []))
      end
    ) (Db.VLAN.get_all ~__context)
  in

  (* Allow pool-join if the host and the pool are on the same management vlan *)
  let assert_management_vlan_are_same () =
    let management_pif = Xapi_host.get_management_interface ~__context ~host:(Helpers.get_localhost ~__context) in
    let vlan_tag = Db.PIF.get_VLAN ~__context ~self:management_pif in
    let remote_management_pif = Client.Host.get_management_interface rpc session_id (get_master ~rpc ~session_id) in
    let remote_vlan_tag = Client.PIF.get_VLAN rpc session_id remote_management_pif in
    if vlan_tag <> remote_vlan_tag then begin
      error "The current host and the pool management vlan does not match: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_management_vlan_does_not_match,
        [Int64.to_string vlan_tag; Int64.to_string remote_vlan_tag]))
    end in

  (* Used to tell XCP and XenServer apart - use PRODUCT_BRAND if present, else use PLATFORM_NAME. *)
  let get_compatibility_name software_version =
    if List.mem_assoc Xapi_globs._product_brand software_version then
      Some (List.assoc Xapi_globs._product_brand software_version)
    else if List.mem_assoc Xapi_globs._platform_name software_version then
      Some (List.assoc Xapi_globs._platform_name software_version)
    else
      None
  in

  let assert_hosts_compatible () =
    let me = Db.Host.get_record ~__context ~self:(Helpers.get_localhost ~__context) in
    let master_ref = get_master rpc session_id in
    let master = Client.Host.get_record ~rpc ~session_id ~self:master_ref in
    let my_software_version = me.API.host_software_version in
    let master_software_version = master.API.host_software_version in
    let compatibility_info x =
      let open Xapi_globs in
      let platform_version = if List.mem_assoc _platform_version x
        then Some (List.assoc _platform_version x)
        else None in
      let compatibility_name = get_compatibility_name x in
      (platform_version, compatibility_name)
    in
    let master_compatibility_info = compatibility_info master_software_version in
    let my_compatibility_info = compatibility_info my_software_version in
    if master_compatibility_info <> my_compatibility_info then begin
      debug "master PLATFORM_VERSION = %s, master compatibility name = %s; my PLATFORM_VERSION = %s, my compatibility name = %s; "
        (Opt.default "Unknown" (fst master_compatibility_info))
        (Opt.default "Unknown" (snd master_compatibility_info))
        (Opt.default "Unknown" (fst my_compatibility_info))
        (Opt.default "Unknown" (snd my_compatibility_info));
      raise (Api_errors.Server_error(Api_errors.pool_hosts_not_compatible, []))
    end in

  let assert_hosts_homogeneous () =
    let me = Helpers.get_localhost ~__context in
    let master_ref = get_master rpc session_id in
    let master = Client.Host.get_record ~rpc ~session_id ~self:master_ref in

    (* Check software version, but as of CA-249786 don't check the build number*)

    let get_software_version_fields fields =
      let open Xapi_globs in
      begin try List.assoc _platform_version fields with _ -> "" end,
      begin match get_compatibility_name fields with Some x -> x | None -> "" end,
      begin try List.assoc _git_id fields with _ -> "" end,
      begin try
          if List.mem_assoc linux_pack_vsn_key fields then "installed"
          else "not present"
        with _ -> "not present" end
    in

    let print_software_version (version,name,id,linux_pack) =
      debug "version:%s, name:%s, id:%s, linux_pack:%s" version name id linux_pack in

    let master_software_version = master.API.host_software_version in
    let my_software_version = Db.Host.get_software_version ~__context ~self:me in

    let my_software_compare = get_software_version_fields my_software_version in
    let master_software_compare = get_software_version_fields master_software_version in

    debug "Pool pre-join Software homogeneity check:";
    debug "Slave software:";
    print_software_version my_software_compare;
    debug "Master software:";
    print_software_version master_software_compare;

    if my_software_compare <> master_software_compare then
      raise (Api_errors.Server_error(Api_errors.pool_hosts_not_homogeneous,["Software version differs"]));

    (* Check CPUs *)

    let my_cpu_vendor = Db.Host.get_cpu_info ~__context ~self:me |> List.assoc "vendor" in
    let pool_cpu_vendor =
      let pool = List.hd (Client.Pool.get_all rpc session_id) in
      Client.Pool.get_cpu_info rpc session_id pool |> List.assoc "vendor"
    in
    debug "Pool pre-join CPU homogeneity check:";
    debug "Slave CPUs: %s" my_cpu_vendor;
    debug "Pool CPUs: %s" pool_cpu_vendor;

    if my_cpu_vendor <> pool_cpu_vendor then
      raise (Api_errors.Server_error(Api_errors.pool_hosts_not_homogeneous,["CPUs differ"])) in

  let assert_not_joining_myself () =
    let master = get_master rpc session_id in
    let master_uuid = Client.Host.get_uuid rpc session_id master in
    let my_uuid = Db.Host.get_uuid ~__context ~self:(Helpers.get_localhost ~__context) in
    if master_uuid = my_uuid then
      let error_msg =
        if 1 < List.length (Db.Host.get_all ~__context)
        then "Host is already part of a pool"
        else "Host cannot become slave of itself" in
      raise (Api_errors.Server_error(Api_errors.operation_not_allowed, [error_msg])) in

  let assert_homogeneous_vswitch_configuration () =
    (* The network backend must be the same as the remote master's backend *)
    let dbg = Context.string_of_task __context in
    let my_backend' = Net.Bridge.get_kind dbg () in
    let my_backend = Network_interface.string_of_kind my_backend' in
    let pool = List.hd (Client.Pool.get_all rpc session_id) in
    let remote_master = Client.Pool.get_master ~rpc ~session_id ~self:pool in
    let remote_masters_backend =
      let v = Client.Host.get_software_version ~rpc ~session_id ~self:remote_master in
      if not (List.mem_assoc "network_backend" v) then
        Network_interface.string_of_kind Network_interface.Bridge
      else
        List.assoc "network_backend" v
    in
    if my_backend <> remote_masters_backend then
      raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["Network backends differ"]));

    match my_backend' with
    | Network_interface.Openvswitch ->
      begin
      let remote_sdn_controllers = Client.SDN_controller.get_all ~rpc ~session_id in
      let my_sdn_controllers = Db.SDN_controller.get_all ~__context in
      (* We assume that each pool has _at most_ one SDN controller *)
      match remote_sdn_controllers, my_sdn_controllers with
      | _, [] -> ()
      | remote_sdn_controller :: _, my_sdn_controller :: _ ->
          (* check that protocol/address/port are identical *)
          let my_sdn_protocol = Db.SDN_controller.get_protocol ~__context ~self:my_sdn_controller in
          let my_sdn_address = Db.SDN_controller.get_address ~__context ~self:my_sdn_controller in
          let my_sdn_port = Db.SDN_controller.get_port ~__context ~self:my_sdn_controller in
          let remote_sdn_protocol = Client.SDN_controller.get_protocol ~rpc ~session_id ~self:remote_sdn_controller in
          let remote_sdn_address = Client.SDN_controller.get_address ~rpc ~session_id ~self:remote_sdn_controller in
          let remote_sdn_port = Client.SDN_controller.get_port ~rpc ~session_id ~self:remote_sdn_controller in
          if my_sdn_protocol <> remote_sdn_protocol
            || my_sdn_address <> remote_sdn_address
            || my_sdn_port <> remote_sdn_port then
            raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["SDN controller differs"]))
      | _ ->
        raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["SDN controller differs"]))
      end
    | _ -> ()
  in

  let assert_homogeneous_primary_address_type () =
    let mgmt_iface = Xapi_host.get_management_interface ~__context ~host:(Helpers.get_localhost ~__context) in
    let mgmt_addr_type = Db.PIF.get_primary_address_type ~__context ~self:mgmt_iface in
    let master = get_master rpc session_id in
    let master_mgmt_iface = Client.Host.get_management_interface rpc session_id master in
    let master_addr_type = Client.PIF.get_primary_address_type rpc session_id master_mgmt_iface in
    if (mgmt_addr_type <> master_addr_type) then
      raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["Primary address type differs"]));
  in

  (* call pre-join asserts *)
  assert_management_interface_exists ();
  ha_is_not_enable_on_me ();
  ha_is_not_enable_on_the_distant_pool ();
  assert_not_joining_myself();
  assert_i_know_of_no_other_hosts();
  assert_no_running_vms_on_me ();
  assert_no_vms_with_current_ops ();
  assert_hosts_compatible ();
  if (not force) then assert_hosts_homogeneous ();
  assert_no_shared_srs_on_me ();
  assert_no_bonds_on_me ();
  assert_no_tunnels_on_me ();
  assert_no_non_management_vlans_on_me ();
  assert_management_vlan_are_same ();
  assert_external_auth_matches ();
  assert_restrictions_match ();
  assert_homogeneous_vswitch_configuration ();
  (* CA-247399: check first the API version and then the database schema *)
  assert_api_version_matches ();
  assert_db_schema_matches ();
  assert_homogeneous_primary_address_type ()

let rec create_or_get_host_on_master __context rpc session_id (host_ref, host) : API.ref_host =
  let my_uuid = host.API.host_uuid in

  let new_host_ref =
    try Client.Host.get_by_uuid rpc session_id my_uuid
    with _ ->
      debug "Found no host with uuid = '%s' on the master, so creating one." my_uuid;

      (* CA-51925: Copy the local cache SR *)
      let my_local_cache_sr = Db.Host.get_local_cache_sr ~__context ~self:host_ref in
      let local_cache_sr = if my_local_cache_sr = Ref.null then Ref.null else
          begin
            let my_local_cache_sr_rec = Db.SR.get_record ~__context ~self:my_local_cache_sr in
            debug "Copying the local cache SR (uuid=%s)" my_local_cache_sr_rec.API.sR_uuid;
            create_or_get_sr_on_master __context rpc session_id (my_local_cache_sr, my_local_cache_sr_rec)
          end in

      (* Look up the value on the master of the pool we are about to join *)
      let master_ssl = Client.Host.get_ssl_legacy ~rpc ~session_id ~self:(get_master rpc session_id) in
      (* Set value in inventory (to control initial behaviour on next xapi start)
         			 * but not in the database of the current pool (the one we're about to leave) *)
      Xapi_inventory.update Xapi_inventory._stunnel_legacy (string_of_bool master_ssl);

      debug "Creating host object on master";
      let ref = Client.Host.create ~rpc ~session_id
          ~uuid:my_uuid
          ~name_label:host.API.host_name_label
          ~name_description:host.API.host_name_description
          ~hostname:host.API.host_hostname
          ~address:host.API.host_address
          ~external_auth_type:host.API.host_external_auth_type
          ~external_auth_service_name:host.API.host_external_auth_service_name
          ~external_auth_configuration:host.API.host_external_auth_configuration
          ~license_params:host.API.host_license_params
          ~edition:host.API.host_edition
          ~license_server:host.API.host_license_server
          (* CA-51925: local_cache_sr can only be written by Host.enable_local_caching_sr but this API
             				 * call is forwarded to the host in question. Since, during a pool-join, the host is offline,
             				 * we need an alternative way of preserving the value of the local_cache_sr field, so it's
             				 * been added to the constructor. *)
          ~local_cache_sr
          ~chipset_info:host.API.host_chipset_info
          ~ssl_legacy:master_ssl
      in

      (* Copy other-config into newly created host record: *)
      no_exn (fun () -> Client.Host.set_other_config ~rpc ~session_id ~self:ref ~value:host.API.host_other_config) ();

      (* Copy the crashdump SR *)
      let my_crashdump_sr = Db.Host.get_crash_dump_sr ~__context ~self:host_ref in
      if my_crashdump_sr <> Ref.null then begin
        let my_crashdump_sr_rec = Db.SR.get_record ~__context ~self:my_crashdump_sr in
        debug "Copying the crashdump SR (uuid=%s)" my_crashdump_sr_rec.API.sR_uuid;
        let crashdump_sr = create_or_get_sr_on_master __context rpc session_id (my_crashdump_sr, my_crashdump_sr_rec) in
        no_exn (fun () -> Client.Host.set_crash_dump_sr ~rpc ~session_id ~self:ref ~value:crashdump_sr) ()
      end;

      (* Copy the suspend image SR *)
      let my_suspend_image_sr = Db.Host.get_crash_dump_sr ~__context ~self:host_ref in
      if my_suspend_image_sr <> Ref.null then begin
        let my_suspend_image_sr_rec = Db.SR.get_record ~__context ~self:my_suspend_image_sr in
        debug "Copying the suspend-image SR (uuid=%s)" my_suspend_image_sr_rec.API.sR_uuid;
        let suspend_image_sr = create_or_get_sr_on_master __context rpc session_id (my_suspend_image_sr, my_suspend_image_sr_rec) in
        no_exn (fun () -> Client.Host.set_suspend_image_sr ~rpc ~session_id ~self:ref ~value:suspend_image_sr) ()
      end;

      ref in

  new_host_ref

and create_or_get_sr_on_master __context rpc session_id (sr_ref, sr) : API.ref_SR =
  let my_uuid = sr.API.sR_uuid in

  let new_sr_ref =
    try Client.SR.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      let my_pbd_ref = List.hd (Db.SR.get_PBDs ~__context ~self:sr_ref) in
      let my_pbd = Db.PBD.get_record ~__context ~self:my_pbd_ref in
      let pbds_on_master = Client.PBD.get_all_records ~rpc ~session_id in

      (* The only possible shared SRs are ISO, as other SRs cannot be shared properly accross pools. *)
      (* In this case, if we find a SR with a PBD having the same device_config field, we pick this SR instead of building a new one *)
      let iso_already_exists_on_master () = List.exists (fun (_,x) -> Listext.List.set_equiv x.API.pBD_device_config my_pbd.API.pBD_device_config) pbds_on_master in
      if sr.API.sR_shared && sr.API.sR_content_type = "iso" && iso_already_exists_on_master () then begin
        let similar_pbd_ref, similar_pbd = List.find (fun (_,x) -> Listext.List.set_equiv x.API.pBD_device_config my_pbd.API.pBD_device_config) pbds_on_master in
        similar_pbd.API.pBD_SR

      end else begin
        debug "Found no SR with uuid = '%s' on the master, so creating one." my_uuid;
        let ref = Client.SR.introduce ~rpc ~session_id
            ~uuid:my_uuid
            ~name_label:sr.API.sR_name_label
            ~name_description:sr.API.sR_name_description
            ~_type:sr.API.sR_type
            ~content_type:sr.API.sR_content_type
            ~shared:false
            ~sm_config:sr.API.sR_sm_config in
        (* copy other-config into newly created sr record: *)
        no_exn (fun () -> Client.SR.set_other_config ~rpc ~session_id ~self:ref ~value:sr.API.sR_other_config) ();
        ref
      end in

  new_sr_ref

let create_or_get_pbd_on_master __context rpc session_id (pbd_ref, pbd) : API.ref_PBD =
  let my_uuid = pbd.API.pBD_uuid in

  let new_pbd_ref =
    try Client.PBD.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      let my_host_ref = pbd.API.pBD_host in
      let my_host = Db.Host.get_record ~__context ~self:my_host_ref in
      let new_host_ref = create_or_get_host_on_master __context rpc session_id (my_host_ref, my_host) in

      let my_sr_ref = pbd.API.pBD_SR in
      let my_sr = Db.SR.get_record ~__context ~self:my_sr_ref in
      let new_sr_ref = create_or_get_sr_on_master __context rpc session_id (my_sr_ref, my_sr) in

      debug "Found no PBD with uuid = '%s' on the master, so creating one." my_uuid;
      Client.PBD.create ~rpc ~session_id
        ~host:new_host_ref
        ~sR:new_sr_ref
        ~other_config:pbd.API.pBD_other_config
        ~device_config:pbd.API.pBD_device_config in

  new_pbd_ref

let create_or_get_vdi_on_master __context rpc session_id (vdi_ref, vdi) : API.ref_VDI =
  let my_uuid = vdi.API.vDI_uuid in
  let my_sr_ref = vdi.API.vDI_SR in
  let my_sr = Db.SR.get_record ~__context ~self:my_sr_ref in

  let new_sr_ref = create_or_get_sr_on_master __context rpc session_id (my_sr_ref, my_sr) in

  let new_vdi_ref =
    try Client.VDI.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      debug "Found no VDI with uuid = '%s' on the master, so creating one." my_uuid;
      Client.VDI.pool_introduce ~rpc ~session_id
        ~uuid:my_uuid
        ~name_label:vdi.API.vDI_name_label
        ~name_description:vdi.API.vDI_name_description
        ~sR:new_sr_ref
        ~_type:vdi.API.vDI_type
        ~sharable:vdi.API.vDI_sharable
        ~read_only:vdi.API.vDI_read_only
        ~other_config:vdi.API.vDI_other_config
        ~location:(Db.VDI.get_location ~__context ~self:vdi_ref)
        ~xenstore_data:vdi.API.vDI_xenstore_data
        ~sm_config:vdi.API.vDI_sm_config
        ~managed:vdi.API.vDI_managed
        ~virtual_size:vdi.API.vDI_virtual_size
        ~physical_utilisation:vdi.API.vDI_physical_utilisation
        ~metadata_of_pool:vdi.API.vDI_metadata_of_pool
        ~is_a_snapshot:vdi.API.vDI_is_a_snapshot
        ~snapshot_time:vdi.API.vDI_snapshot_time
        ~snapshot_of:vdi.API.vDI_snapshot_of in
  new_vdi_ref

let create_or_get_network_on_master __context rpc session_id (network_ref, network) : API.ref_network =
  let my_bridge = network.API.network_bridge in
  let is_physical = match network.API.network_PIFs with
    | [] -> false
    | hd :: _ -> Db.PIF.get_physical ~__context ~self:hd
  in
  let is_himn =
    (List.mem_assoc Xapi_globs.is_host_internal_management_network network.API.network_other_config) &&
    (List.assoc Xapi_globs.is_host_internal_management_network network.API.network_other_config = "true")
  in

  let new_network_ref =
    if is_physical || is_himn then
      (* Physical network or Host Internal Management Network:
         			 * try to join an existing network with the same bridge name, or create one.
         			 * This relies on the convention that physical PIFs with the same device name need to be connected.
         			 * Furthermore, there should be only one Host Internal Management Network in a pool. *)
      try
        let pool_networks = Client.Network.get_all_records ~rpc ~session_id in
        let net_ref, _ = List.find (fun (_, net) -> net.API.network_bridge = my_bridge) pool_networks in
        net_ref
      with _ ->
        debug "Found no network with bridge = '%s' on the master, so creating one." my_bridge;
        Client.Network.pool_introduce ~rpc ~session_id
          ~name_label:network.API.network_name_label
          ~name_description:network.API.network_name_description
          ~mTU:network.API.network_MTU
          ~other_config:network.API.network_other_config
          ~bridge:network.API.network_bridge
          ~managed:network.API.network_managed
    else begin
      debug "Recreating network '%s' as internal network." network.API.network_name_label;
      (* This call will generate a new 'xapi#' bridge name rather than keeping the
         			 * current, possibly colliding one. *)
      Client.Network.create ~rpc ~session_id
        ~name_label:network.API.network_name_label
        ~name_description:network.API.network_name_description
        ~mTU:network.API.network_MTU
        ~other_config:network.API.network_other_config
        ~bridge:network.API.network_bridge
        ~managed:network.API.network_managed
        ~tags:network.API.network_tags
    end
  in

  new_network_ref

let create_or_get_pif_on_master __context rpc session_id (pif_ref, pif) : API.ref_PIF =
  let my_uuid = pif.API.pIF_uuid in

  let my_host_ref = pif.API.pIF_host in
  let my_host = Db.Host.get_record ~__context ~self:my_host_ref in
  let new_host_ref = create_or_get_host_on_master __context rpc session_id (my_host_ref, my_host) in

  let new_network_ref =
    if pif.API.pIF_VLAN <> -1L then begin
      (* Get the remote management network for management VLAN PIF *)
      let remote_mgmt_pif = Client.Host.get_management_interface rpc session_id (get_master ~rpc ~session_id) in
      Client.PIF.get_network rpc session_id remote_mgmt_pif
    end
    else begin
      let my_network_ref = pif.API.pIF_network in
      let my_network = Db.Network.get_record ~__context ~self:my_network_ref in
      create_or_get_network_on_master __context rpc session_id (my_network_ref, my_network)
    end
  in

  let new_pif_ref =
    try Client.PIF.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      debug "Found no PIF with uuid = '%s' on the master, so creating one." my_uuid;
      Client.PIF.pool_introduce ~rpc ~session_id
        ~device:pif.API.pIF_device
        ~network:new_network_ref
        ~host:new_host_ref
        ~mAC:pif.API.pIF_MAC
        ~mTU:pif.API.pIF_MTU
        ~vLAN:pif.API.pIF_VLAN
        ~physical:pif.API.pIF_physical
        ~ip_configuration_mode:pif.API.pIF_ip_configuration_mode
        ~iP:pif.API.pIF_IP
        ~netmask:pif.API.pIF_netmask
        ~gateway:pif.API.pIF_gateway
        ~dNS:pif.API.pIF_DNS
        ~bond_slave_of:pif.API.pIF_bond_slave_of
        ~vLAN_master_of:pif.API.pIF_VLAN_master_of
        ~management:pif.API.pIF_management
        ~other_config:pif.API.pIF_other_config
        ~disallow_unplug:pif.API.pIF_disallow_unplug
        ~ipv6_configuration_mode:pif.API.pIF_ipv6_configuration_mode
        ~iPv6:pif.API.pIF_IPv6
        ~ipv6_gateway:pif.API.pIF_ipv6_gateway
        ~primary_address_type:pif.API.pIF_primary_address_type
        ~managed:pif.API.pIF_managed
        ~properties:pif.API.pIF_properties in

  new_pif_ref

let create_or_get_pvs_site_on_master __context rpc session_id (pvs_site_ref, pvs_site) : API.ref_PVS_site =
  let my_pvs_uuid = pvs_site.API.pVS_site_PVS_uuid in
  let new_pvs_site_ref =
    let expr = "field \"PVS_uuid\"=\"" ^ my_pvs_uuid ^ "\"" in
    match Client.PVS_site.get_all_records_where ~rpc ~session_id ~expr with
    | [] ->
      debug "Found no PVS site with PVS_uuid = '%s' on the master, so creating one." my_pvs_uuid;
      let new_pvs_site = Client.PVS_site.introduce ~rpc ~session_id
          ~name_label:pvs_site.API.pVS_site_name_label
          ~name_description:pvs_site.API.pVS_site_name_description
          ~pVS_uuid:pvs_site.API.pVS_site_PVS_uuid in
      (* Update PVS servers *)
      let (_ : API.ref_PVS_server list) = List.map (fun pvs_server ->
          let pvs_record = Db.PVS_server.get_record ~__context ~self:pvs_server in
          Client.PVS_server.introduce ~rpc ~session_id
            ~addresses:pvs_record.API.pVS_server_addresses
            ~first_port:pvs_record.API.pVS_server_first_port
            ~last_port:pvs_record.API.pVS_server_last_port
            ~site:new_pvs_site
        ) (Db.PVS_site.get_servers ~__context ~self:pvs_site_ref) in
      new_pvs_site
    | (pvs_site, _) :: _ -> pvs_site
  in
  new_pvs_site_ref

let create_or_get_pvs_cache_storage_on_master __context rpc session_id (pcs_ref, pcs) : API.ref_PVS_cache_storage =
  let my_host_ref = pcs.API.pVS_cache_storage_host in
  let my_host = Db.Host.get_record ~__context ~self:my_host_ref in
  let new_host_ref = create_or_get_host_on_master __context rpc session_id (my_host_ref, my_host) in

  let my_sr_ref = pcs.API.pVS_cache_storage_SR in
  let my_sr = Db.SR.get_record ~__context ~self:my_sr_ref in
  let new_sr_ref = create_or_get_sr_on_master __context rpc session_id (my_sr_ref, my_sr) in

  let my_pvs_site_ref = pcs.API.pVS_cache_storage_site in
  let my_pvs_site = Db.PVS_site.get_record ~__context ~self:my_pvs_site_ref in
  let new_pvs_site_ref = create_or_get_pvs_site_on_master __context rpc session_id (my_pvs_site_ref, my_pvs_site) in

  let new_pvs_cache_storage_ref =
    let expr =
      Printf.sprintf "field \"site\"=\"%s\" and field \"host\"=\"%s\""
        (Ref.string_of new_pvs_site_ref)
        (Ref.string_of new_host_ref)
    in
    match Client.PVS_cache_storage.get_all_records_where ~rpc ~session_id ~expr with
    | [] ->
      Client.PVS_cache_storage.create ~rpc ~session_id
        ~host:new_host_ref
        ~sR:new_sr_ref
        ~site:new_pvs_site_ref
        ~size:pcs.API.pVS_cache_storage_size
    | (pvs_cache_storage, _) :: _ -> pvs_cache_storage
  in
  new_pvs_cache_storage_ref

let create_or_get_secret_on_master __context rpc session_id (secret_ref, secret) : API.ref_secret =
  let my_uuid = secret.API.secret_uuid in
  let my_value = secret.API.secret_value in
  let new_secret_ref =
    try Client.Secret.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      debug "Found no secret with uuid = '%s' on master, so creating one." my_uuid;
      Client.Secret.introduce ~rpc ~session_id ~uuid:my_uuid ~value:my_value ~other_config:[]
  in
  new_secret_ref

let protect_exn f x =
  try Some (f x)
  with e ->
    debug "Ignoring exception: %s" (Printexc.to_string e);
    Debug.log_backtrace e (Backtrace.get e);
    None

(* Remark: the order in which we create the object in the distant database is not very important, as we have *)
(* an unique way to identify each object and thus we know if we need to create them or if it is already done *)
let update_non_vm_metadata ~__context ~rpc ~session_id =

  (* Update hosts *)
  let my_hosts = Db.Host.get_all_records ~__context in
  let (_ : API.ref_host option list) =
    List.map (protect_exn (create_or_get_host_on_master __context rpc session_id)) my_hosts in

  (* Update SRs *)
  let my_srs = Db.SR.get_all_records ~__context in
  let (_ : API.ref_SR option list) =
    List.map (protect_exn (create_or_get_sr_on_master __context rpc session_id)) my_srs in

  (* Update PBDs *)
  let my_pbds = Db.PBD.get_all_records ~__context in
  let (_ : API.ref_PBD option list) =
    List.map (protect_exn (create_or_get_pbd_on_master __context rpc session_id)) my_pbds in

  (* Update VDIs *)
  let my_vdis = Db.VDI.get_all_records ~__context in
  let (_ : API.ref_VDI option list) =
    List.map (protect_exn (create_or_get_vdi_on_master __context rpc session_id)) my_vdis in

  (* Update networks *)
  let my_networks = Db.Network.get_all_records ~__context in
  let (_ : API.ref_network option list) =
    List.map (protect_exn (create_or_get_network_on_master __context rpc session_id)) my_networks in

  (* update PIFs *)
  let my_pifs = Db.PIF.get_records_where ~__context ~expr:(
      Eq (Field "physical", Literal "true")
    ) in
  let (_ : API.ref_PIF option list) =
    List.map (protect_exn (create_or_get_pif_on_master __context rpc session_id)) my_pifs in

  (* update PVS sites *)
  let my_pvs_sites = Db.PVS_site.get_all_records ~__context in
  let (_ : API.ref_PVS_site option list) =
    List.map (protect_exn (create_or_get_pvs_site_on_master __context rpc session_id)) my_pvs_sites in

  (* update PVS_cache_storage *)
  let my_pvs_cache_storages = Db.PVS_cache_storage.get_all_records ~__context in
  let (_ : API.ref_PVS_cache_storage option list) =<commit_msg>CP-20162: Create Management VLAN on Pool.join<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.
 *)

open Stdext
open Client
open Db_filter_types
open Pervasiveext
open Threadext
open Xstringext
open Listext
open Network

module L = Debug.Make(struct let name="license" end)
module D=Debug.Make(struct let name="xapi" end)
open D
open Workload_balancing

(* Surpress exceptions *)
let no_exn f x =
  try ignore (f x)
  with exn ->
    debug "Ignoring exception: %s" (ExnHelper.string_of_exn exn)

let rpc host_address xml =
  try
    Helpers.make_remote_rpc host_address xml
  with Xmlrpc_client.Connection_reset ->
    raise (Api_errors.Server_error(Api_errors.pool_joining_host_connection_failed, []))

let get_master ~rpc ~session_id =
  let pool = List.hd (Client.Pool.get_all rpc session_id) in
  Client.Pool.get_master rpc session_id pool

(* Pre-join asserts *)
let pre_join_checks ~__context ~rpc ~session_id ~force =
  (* I cannot join a Pool unless my management interface exists in the db, otherwise
     	   Pool.eject will fail to rewrite network interface files. *)
  let assert_management_interface_exists () =
    try
      let (_: API.ref_PIF) = Xapi_host.get_management_interface ~__context ~host:(Helpers.get_localhost ~__context) in
      ()
    with _ ->
      error "Pool.join/Pool.eject requires a properly configured management interface. Wait for xapi/firstboot initialisation to complete and then retry.";
      raise (Api_errors.Server_error(Api_errors.host_still_booting, [])) in

  (* I cannot join a Pool if I have HA already enabled on me *)
  let ha_is_not_enable_on_me () =
    let pool = Helpers.get_pool ~__context in
    if Db.Pool.get_ha_enabled ~__context ~self:pool then begin
      error "Cannot join pool as HA is enabled";
      raise (Api_errors.Server_error(Api_errors.ha_is_enabled, []))
    end in

  (* I Cannot join a Pool if it has HA enabled on it *)
  let ha_is_not_enable_on_the_distant_pool () =
    let pool = List.hd (Client.Pool.get_all rpc session_id) in
    if Client.Pool.get_ha_enabled rpc session_id pool then begin
      error "Cannot join pool which already has HA enabled";
      raise (Api_errors.Server_error(Api_errors.ha_is_enabled, []));
    end in

  (* CA-26975: Pool edition MUST match *)
  let assert_restrictions_match () =
    let my_edition = Db.Host.get_edition ~__context ~self:(Helpers.get_localhost ~__context) in
    let host_records = List.map snd (Client.Host.get_all_records ~rpc ~session_id) in
    let pool_editions = List.map (fun host_r -> host_r.API.host_edition) host_records in
    (* If all hosts have the same edition string, we need do no more. *)
    if List.fold_left (fun b edn -> edn = my_edition && b) true pool_editions then ()
    else try
        (* We have different edition strings so must consult v6d for their significance.
           			 * This will fail with v6d_failure if v6d is not running. *)
        let editions = V6_client.get_editions "assert_restrictions_match" in
        let edition_to_int e =
          try
            match List.find (fun (name, _) -> name = e) editions with _, (_, _, a) -> a
          with Not_found ->
            (* Happens if pool has edition "free/libre" (no v6d) *)
            error "Pool.join failed: pool has a host with edition unknown to v6d: %s" e;
            raise (Api_errors.Server_error(Api_errors.license_host_pool_mismatch,
                                           ["Edition \"" ^ e ^ "\" from pool is not known to v6d."]))
        in
        let min_edition l =
          List.fold_left (fun m e -> if edition_to_int e < edition_to_int m then e else m) (List.hd l) l
        in
        (* get pool edition: the "minimum" edition among all hosts *)
        let pool_edition = min_edition pool_editions in
        (* compare my edition to pool edition *)
        if (edition_to_int pool_edition) <> (edition_to_int my_edition) then begin
          error "Pool.join failed due to edition mismatch";
          error "Remote has %s" pool_edition;
          error "Local has  %s" my_edition;
          raise (Api_errors.Server_error(Api_errors.license_host_pool_mismatch,
                                         ["host edition = \""^my_edition^"\""; "pool edition = \""^pool_edition^"\""]))
        end
      with Api_errors.Server_error (code, []) when code = Api_errors.v6d_failure ->
        error "Pool.join failed because edition strings differ and local has no license daemon running.";
        let pool_edn_list_str = "[" ^ (String.concat "; " pool_editions) ^ "]" in
        error "Remote editions: %s" pool_edn_list_str;
        error "Local edition: %s" my_edition;
        raise (Api_errors.Server_error (code, ["The pool uses v6d. Pool edition list = " ^ pool_edn_list_str]))
  in

  let assert_api_version_matches () =
    let master = get_master rpc session_id in
    let candidate_slave = Helpers.get_localhost ~__context in
    let master_major = Client.Host.get_API_version_major ~rpc ~session_id ~self:master in
    let master_minor = Client.Host.get_API_version_minor ~rpc ~session_id ~self:master in
    let slave_major = Db.Host.get_API_version_major ~__context ~self:candidate_slave in
    let slave_minor = Db.Host.get_API_version_minor ~__context ~self:candidate_slave in
    if master_major <> slave_major || master_minor <> slave_minor then
    begin
      error "The joining host's API version is %Ld.%Ld while the master's is %Ld.%Ld"
        slave_major slave_minor master_major master_minor;
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_must_have_same_api_version,
        [Printf.sprintf "%Ld.%Ld" slave_major slave_minor; Printf.sprintf "%Ld.%Ld" master_major master_minor;]))
    end
  in

  let assert_db_schema_matches () =
    let master = get_master rpc session_id in
    let candidate_slave = Helpers.get_localhost ~__context in
    let master_sw_version = Client.Host.get_software_version ~rpc ~session_id ~self:master in
    let slave_sw_version = Db.Host.get_software_version ~__context ~self:candidate_slave in
    let master_db_schema = try List.assoc Xapi_globs._db_schema master_sw_version with _ -> "" in
    let slave_db_schema = try List.assoc Xapi_globs._db_schema slave_sw_version with _ -> "" in
    if master_db_schema = "" || slave_db_schema = "" ||  master_db_schema <> slave_db_schema then
    begin
        error "The joining host's database schema is %s; the master's is %s"
          slave_db_schema master_db_schema;
        raise (Api_errors.Server_error(Api_errors.pool_joining_host_must_have_same_db_schema,
          [slave_db_schema; master_db_schema]))
    end
  in

  (* CP-700: Restrict pool.join if AD configuration of slave-to-be does not match *)
  (* that of master of pool-to-join *)
  let assert_external_auth_matches () =
    let master = get_master rpc session_id in
    let slavetobe = Helpers.get_localhost ~__context in
    let slavetobe_auth_type = Db.Host.get_external_auth_type ~__context ~self:slavetobe in
    let slavetobe_auth_service_name = Db.Host.get_external_auth_service_name ~__context ~self:slavetobe in
    let master_auth_type = Client.Host.get_external_auth_type ~rpc ~session_id ~self:master in
    let master_auth_service_name = Client.Host.get_external_auth_service_name ~rpc ~session_id ~self:master in
    debug "Verifying if external auth configuration of master %s (auth_type=%s service_name=%s) matches that of slave-to-be %s (auth-type=%s service_name=%s)"
      (Client.Host.get_name_label ~rpc ~session_id ~self:master) master_auth_type master_auth_service_name
      (Db.Host.get_name_label ~__context ~self:slavetobe) slavetobe_auth_type slavetobe_auth_service_name;
    if (slavetobe_auth_type <> master_auth_type)
    || (slavetobe_auth_service_name <> master_auth_service_name) then begin
      error "Cannot join pool whose external authentication configuration is different";
      raise (Api_errors.Server_error(Api_errors.pool_joining_external_auth_mismatch, []))
    end in

  let assert_i_know_of_no_other_hosts () =
    let hosts = Db.Host.get_all ~__context in
    if List.length hosts > 1 then begin
      error "The current host is already the master of other hosts: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_be_master_of_other_hosts, []))
    end in

  let assert_no_running_vms_on_me () =
    let my_vms = Db.VM.get_all_records ~__context in
    let my_running_vms =
      List.filter
        (fun (_,vmrec) -> (not (Helpers.is_domain_zero ~__context (Db.VM.get_by_uuid ~__context ~uuid:vmrec.API.vM_uuid)))
                          && vmrec.API.vM_power_state = `Running) my_vms in
    if List.length my_running_vms > 0 then begin
      error "The current host has running or suspended VMs: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_have_running_VMs, []))
    end in

  let assert_no_vms_with_current_ops () =
    let my_vms = Db.VM.get_all_records ~__context in
    let vms_with_current_ops =
      List.filter (fun (_,vmr) -> (List.length vmr.API.vM_current_operations)>0 ) my_vms in
    if List.length vms_with_current_ops > 0 then begin
      error "The current host has VMs with current operations: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_have_vms_with_current_operations, []))
    end in

  let assert_no_shared_srs_on_me () =
    let my_srs = Db.SR.get_all_records ~__context in
    let my_shared_srs = List.filter (fun (sr,srec)-> srec.API.sR_shared && not srec.API.sR_is_tools_sr) my_srs in
    if not (my_shared_srs = []) then begin
      error "The current host has shared SRs: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_cannot_contain_shared_SRs, []))
    end in

  (* Allow pool-join if host does not have any bonds *)
  let assert_no_bonds_on_me () =
    if Db.Bond.get_all ~__context <> [] then begin
      error "The current host has network bonds: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_has_bonds, []))
    end in

  (* Allow pool-join if host does not have any tunnels *)
  let assert_no_tunnels_on_me () =
    if Db.Tunnel.get_all ~__context <> [] then begin
      error "The current host has tunnels: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_has_tunnels, []))
    end in

  (* Allow pool-join if host does not have any non-management VLANs *)
  let assert_no_non_management_vlans_on_me () =
    List.iter (fun self ->
      let pif = Db.VLAN.get_untagged_PIF ~__context ~self in
      if Db.PIF.get_management ~__context ~self:pif <> true then begin
        error "The current host has non-management vlans: it cannot join a new pool";
        raise (Api_errors.Server_error(Api_errors.pool_joining_host_has_non_management_vlans, []))
      end
    ) (Db.VLAN.get_all ~__context)
  in

  (* Allow pool-join if the host and the pool are on the same management vlan *)
  let assert_management_vlan_are_same () =
    let management_pif = Xapi_host.get_management_interface ~__context ~host:(Helpers.get_localhost ~__context) in
    let vlan_tag = Db.PIF.get_VLAN ~__context ~self:management_pif in
    let remote_management_pif = Client.Host.get_management_interface rpc session_id (get_master ~rpc ~session_id) in
    let remote_vlan_tag = Client.PIF.get_VLAN rpc session_id remote_management_pif in
    if vlan_tag <> remote_vlan_tag then begin
      error "The current host and the pool management vlan does not match: it cannot join a new pool";
      raise (Api_errors.Server_error(Api_errors.pool_joining_host_management_vlan_does_not_match,
        [Int64.to_string vlan_tag; Int64.to_string remote_vlan_tag]))
    end in

  (* Used to tell XCP and XenServer apart - use PRODUCT_BRAND if present, else use PLATFORM_NAME. *)
  let get_compatibility_name software_version =
    if List.mem_assoc Xapi_globs._product_brand software_version then
      Some (List.assoc Xapi_globs._product_brand software_version)
    else if List.mem_assoc Xapi_globs._platform_name software_version then
      Some (List.assoc Xapi_globs._platform_name software_version)
    else
      None
  in

  let assert_hosts_compatible () =
    let me = Db.Host.get_record ~__context ~self:(Helpers.get_localhost ~__context) in
    let master_ref = get_master rpc session_id in
    let master = Client.Host.get_record ~rpc ~session_id ~self:master_ref in
    let my_software_version = me.API.host_software_version in
    let master_software_version = master.API.host_software_version in
    let compatibility_info x =
      let open Xapi_globs in
      let platform_version = if List.mem_assoc _platform_version x
        then Some (List.assoc _platform_version x)
        else None in
      let compatibility_name = get_compatibility_name x in
      (platform_version, compatibility_name)
    in
    let master_compatibility_info = compatibility_info master_software_version in
    let my_compatibility_info = compatibility_info my_software_version in
    if master_compatibility_info <> my_compatibility_info then begin
      debug "master PLATFORM_VERSION = %s, master compatibility name = %s; my PLATFORM_VERSION = %s, my compatibility name = %s; "
        (Opt.default "Unknown" (fst master_compatibility_info))
        (Opt.default "Unknown" (snd master_compatibility_info))
        (Opt.default "Unknown" (fst my_compatibility_info))
        (Opt.default "Unknown" (snd my_compatibility_info));
      raise (Api_errors.Server_error(Api_errors.pool_hosts_not_compatible, []))
    end in

  let assert_hosts_homogeneous () =
    let me = Helpers.get_localhost ~__context in
    let master_ref = get_master rpc session_id in
    let master = Client.Host.get_record ~rpc ~session_id ~self:master_ref in

    (* Check software version, but as of CA-249786 don't check the build number*)

    let get_software_version_fields fields =
      let open Xapi_globs in
      begin try List.assoc _platform_version fields with _ -> "" end,
      begin match get_compatibility_name fields with Some x -> x | None -> "" end,
      begin try List.assoc _git_id fields with _ -> "" end,
      begin try
          if List.mem_assoc linux_pack_vsn_key fields then "installed"
          else "not present"
        with _ -> "not present" end
    in

    let print_software_version (version,name,id,linux_pack) =
      debug "version:%s, name:%s, id:%s, linux_pack:%s" version name id linux_pack in

    let master_software_version = master.API.host_software_version in
    let my_software_version = Db.Host.get_software_version ~__context ~self:me in

    let my_software_compare = get_software_version_fields my_software_version in
    let master_software_compare = get_software_version_fields master_software_version in

    debug "Pool pre-join Software homogeneity check:";
    debug "Slave software:";
    print_software_version my_software_compare;
    debug "Master software:";
    print_software_version master_software_compare;

    if my_software_compare <> master_software_compare then
      raise (Api_errors.Server_error(Api_errors.pool_hosts_not_homogeneous,["Software version differs"]));

    (* Check CPUs *)

    let my_cpu_vendor = Db.Host.get_cpu_info ~__context ~self:me |> List.assoc "vendor" in
    let pool_cpu_vendor =
      let pool = List.hd (Client.Pool.get_all rpc session_id) in
      Client.Pool.get_cpu_info rpc session_id pool |> List.assoc "vendor"
    in
    debug "Pool pre-join CPU homogeneity check:";
    debug "Slave CPUs: %s" my_cpu_vendor;
    debug "Pool CPUs: %s" pool_cpu_vendor;

    if my_cpu_vendor <> pool_cpu_vendor then
      raise (Api_errors.Server_error(Api_errors.pool_hosts_not_homogeneous,["CPUs differ"])) in

  let assert_not_joining_myself () =
    let master = get_master rpc session_id in
    let master_uuid = Client.Host.get_uuid rpc session_id master in
    let my_uuid = Db.Host.get_uuid ~__context ~self:(Helpers.get_localhost ~__context) in
    if master_uuid = my_uuid then
      let error_msg =
        if 1 < List.length (Db.Host.get_all ~__context)
        then "Host is already part of a pool"
        else "Host cannot become slave of itself" in
      raise (Api_errors.Server_error(Api_errors.operation_not_allowed, [error_msg])) in

  let assert_homogeneous_vswitch_configuration () =
    (* The network backend must be the same as the remote master's backend *)
    let dbg = Context.string_of_task __context in
    let my_backend' = Net.Bridge.get_kind dbg () in
    let my_backend = Network_interface.string_of_kind my_backend' in
    let pool = List.hd (Client.Pool.get_all rpc session_id) in
    let remote_master = Client.Pool.get_master ~rpc ~session_id ~self:pool in
    let remote_masters_backend =
      let v = Client.Host.get_software_version ~rpc ~session_id ~self:remote_master in
      if not (List.mem_assoc "network_backend" v) then
        Network_interface.string_of_kind Network_interface.Bridge
      else
        List.assoc "network_backend" v
    in
    if my_backend <> remote_masters_backend then
      raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["Network backends differ"]));

    match my_backend' with
    | Network_interface.Openvswitch ->
      begin
      let remote_sdn_controllers = Client.SDN_controller.get_all ~rpc ~session_id in
      let my_sdn_controllers = Db.SDN_controller.get_all ~__context in
      (* We assume that each pool has _at most_ one SDN controller *)
      match remote_sdn_controllers, my_sdn_controllers with
      | _, [] -> ()
      | remote_sdn_controller :: _, my_sdn_controller :: _ ->
          (* check that protocol/address/port are identical *)
          let my_sdn_protocol = Db.SDN_controller.get_protocol ~__context ~self:my_sdn_controller in
          let my_sdn_address = Db.SDN_controller.get_address ~__context ~self:my_sdn_controller in
          let my_sdn_port = Db.SDN_controller.get_port ~__context ~self:my_sdn_controller in
          let remote_sdn_protocol = Client.SDN_controller.get_protocol ~rpc ~session_id ~self:remote_sdn_controller in
          let remote_sdn_address = Client.SDN_controller.get_address ~rpc ~session_id ~self:remote_sdn_controller in
          let remote_sdn_port = Client.SDN_controller.get_port ~rpc ~session_id ~self:remote_sdn_controller in
          if my_sdn_protocol <> remote_sdn_protocol
            || my_sdn_address <> remote_sdn_address
            || my_sdn_port <> remote_sdn_port then
            raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["SDN controller differs"]))
      | _ ->
        raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["SDN controller differs"]))
      end
    | _ -> ()
  in

  let assert_homogeneous_primary_address_type () =
    let mgmt_iface = Xapi_host.get_management_interface ~__context ~host:(Helpers.get_localhost ~__context) in
    let mgmt_addr_type = Db.PIF.get_primary_address_type ~__context ~self:mgmt_iface in
    let master = get_master rpc session_id in
    let master_mgmt_iface = Client.Host.get_management_interface rpc session_id master in
    let master_addr_type = Client.PIF.get_primary_address_type rpc session_id master_mgmt_iface in
    if (mgmt_addr_type <> master_addr_type) then
      raise (Api_errors.Server_error(Api_errors.operation_not_allowed, ["Primary address type differs"]));
  in

  (* call pre-join asserts *)
  assert_management_interface_exists ();
  ha_is_not_enable_on_me ();
  ha_is_not_enable_on_the_distant_pool ();
  assert_not_joining_myself();
  assert_i_know_of_no_other_hosts();
  assert_no_running_vms_on_me ();
  assert_no_vms_with_current_ops ();
  assert_hosts_compatible ();
  if (not force) then assert_hosts_homogeneous ();
  assert_no_shared_srs_on_me ();
  assert_no_bonds_on_me ();
  assert_no_tunnels_on_me ();
  assert_no_non_management_vlans_on_me ();
  assert_management_vlan_are_same ();
  assert_external_auth_matches ();
  assert_restrictions_match ();
  assert_homogeneous_vswitch_configuration ();
  (* CA-247399: check first the API version and then the database schema *)
  assert_api_version_matches ();
  assert_db_schema_matches ();
  assert_homogeneous_primary_address_type ()

let rec create_or_get_host_on_master __context rpc session_id (host_ref, host) : API.ref_host =
  let my_uuid = host.API.host_uuid in

  let new_host_ref =
    try Client.Host.get_by_uuid rpc session_id my_uuid
    with _ ->
      debug "Found no host with uuid = '%s' on the master, so creating one." my_uuid;

      (* CA-51925: Copy the local cache SR *)
      let my_local_cache_sr = Db.Host.get_local_cache_sr ~__context ~self:host_ref in
      let local_cache_sr = if my_local_cache_sr = Ref.null then Ref.null else
          begin
            let my_local_cache_sr_rec = Db.SR.get_record ~__context ~self:my_local_cache_sr in
            debug "Copying the local cache SR (uuid=%s)" my_local_cache_sr_rec.API.sR_uuid;
            create_or_get_sr_on_master __context rpc session_id (my_local_cache_sr, my_local_cache_sr_rec)
          end in

      (* Look up the value on the master of the pool we are about to join *)
      let master_ssl = Client.Host.get_ssl_legacy ~rpc ~session_id ~self:(get_master rpc session_id) in
      (* Set value in inventory (to control initial behaviour on next xapi start)
         			 * but not in the database of the current pool (the one we're about to leave) *)
      Xapi_inventory.update Xapi_inventory._stunnel_legacy (string_of_bool master_ssl);

      debug "Creating host object on master";
      let ref = Client.Host.create ~rpc ~session_id
          ~uuid:my_uuid
          ~name_label:host.API.host_name_label
          ~name_description:host.API.host_name_description
          ~hostname:host.API.host_hostname
          ~address:host.API.host_address
          ~external_auth_type:host.API.host_external_auth_type
          ~external_auth_service_name:host.API.host_external_auth_service_name
          ~external_auth_configuration:host.API.host_external_auth_configuration
          ~license_params:host.API.host_license_params
          ~edition:host.API.host_edition
          ~license_server:host.API.host_license_server
          (* CA-51925: local_cache_sr can only be written by Host.enable_local_caching_sr but this API
             				 * call is forwarded to the host in question. Since, during a pool-join, the host is offline,
             				 * we need an alternative way of preserving the value of the local_cache_sr field, so it's
             				 * been added to the constructor. *)
          ~local_cache_sr
          ~chipset_info:host.API.host_chipset_info
          ~ssl_legacy:master_ssl
      in

      (* Copy other-config into newly created host record: *)
      no_exn (fun () -> Client.Host.set_other_config ~rpc ~session_id ~self:ref ~value:host.API.host_other_config) ();

      (* Copy the crashdump SR *)
      let my_crashdump_sr = Db.Host.get_crash_dump_sr ~__context ~self:host_ref in
      if my_crashdump_sr <> Ref.null then begin
        let my_crashdump_sr_rec = Db.SR.get_record ~__context ~self:my_crashdump_sr in
        debug "Copying the crashdump SR (uuid=%s)" my_crashdump_sr_rec.API.sR_uuid;
        let crashdump_sr = create_or_get_sr_on_master __context rpc session_id (my_crashdump_sr, my_crashdump_sr_rec) in
        no_exn (fun () -> Client.Host.set_crash_dump_sr ~rpc ~session_id ~self:ref ~value:crashdump_sr) ()
      end;

      (* Copy the suspend image SR *)
      let my_suspend_image_sr = Db.Host.get_crash_dump_sr ~__context ~self:host_ref in
      if my_suspend_image_sr <> Ref.null then begin
        let my_suspend_image_sr_rec = Db.SR.get_record ~__context ~self:my_suspend_image_sr in
        debug "Copying the suspend-image SR (uuid=%s)" my_suspend_image_sr_rec.API.sR_uuid;
        let suspend_image_sr = create_or_get_sr_on_master __context rpc session_id (my_suspend_image_sr, my_suspend_image_sr_rec) in
        no_exn (fun () -> Client.Host.set_suspend_image_sr ~rpc ~session_id ~self:ref ~value:suspend_image_sr) ()
      end;

      ref in

  new_host_ref

and create_or_get_sr_on_master __context rpc session_id (sr_ref, sr) : API.ref_SR =
  let my_uuid = sr.API.sR_uuid in

  let new_sr_ref =
    try Client.SR.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      let my_pbd_ref = List.hd (Db.SR.get_PBDs ~__context ~self:sr_ref) in
      let my_pbd = Db.PBD.get_record ~__context ~self:my_pbd_ref in
      let pbds_on_master = Client.PBD.get_all_records ~rpc ~session_id in

      (* The only possible shared SRs are ISO, as other SRs cannot be shared properly accross pools. *)
      (* In this case, if we find a SR with a PBD having the same device_config field, we pick this SR instead of building a new one *)
      let iso_already_exists_on_master () = List.exists (fun (_,x) -> Listext.List.set_equiv x.API.pBD_device_config my_pbd.API.pBD_device_config) pbds_on_master in
      if sr.API.sR_shared && sr.API.sR_content_type = "iso" && iso_already_exists_on_master () then begin
        let similar_pbd_ref, similar_pbd = List.find (fun (_,x) -> Listext.List.set_equiv x.API.pBD_device_config my_pbd.API.pBD_device_config) pbds_on_master in
        similar_pbd.API.pBD_SR

      end else begin
        debug "Found no SR with uuid = '%s' on the master, so creating one." my_uuid;
        let ref = Client.SR.introduce ~rpc ~session_id
            ~uuid:my_uuid
            ~name_label:sr.API.sR_name_label
            ~name_description:sr.API.sR_name_description
            ~_type:sr.API.sR_type
            ~content_type:sr.API.sR_content_type
            ~shared:false
            ~sm_config:sr.API.sR_sm_config in
        (* copy other-config into newly created sr record: *)
        no_exn (fun () -> Client.SR.set_other_config ~rpc ~session_id ~self:ref ~value:sr.API.sR_other_config) ();
        ref
      end in

  new_sr_ref

let create_or_get_pbd_on_master __context rpc session_id (pbd_ref, pbd) : API.ref_PBD =
  let my_uuid = pbd.API.pBD_uuid in

  let new_pbd_ref =
    try Client.PBD.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      let my_host_ref = pbd.API.pBD_host in
      let my_host = Db.Host.get_record ~__context ~self:my_host_ref in
      let new_host_ref = create_or_get_host_on_master __context rpc session_id (my_host_ref, my_host) in

      let my_sr_ref = pbd.API.pBD_SR in
      let my_sr = Db.SR.get_record ~__context ~self:my_sr_ref in
      let new_sr_ref = create_or_get_sr_on_master __context rpc session_id (my_sr_ref, my_sr) in

      debug "Found no PBD with uuid = '%s' on the master, so creating one." my_uuid;
      Client.PBD.create ~rpc ~session_id
        ~host:new_host_ref
        ~sR:new_sr_ref
        ~other_config:pbd.API.pBD_other_config
        ~device_config:pbd.API.pBD_device_config in

  new_pbd_ref

let create_or_get_vdi_on_master __context rpc session_id (vdi_ref, vdi) : API.ref_VDI =
  let my_uuid = vdi.API.vDI_uuid in
  let my_sr_ref = vdi.API.vDI_SR in
  let my_sr = Db.SR.get_record ~__context ~self:my_sr_ref in

  let new_sr_ref = create_or_get_sr_on_master __context rpc session_id (my_sr_ref, my_sr) in

  let new_vdi_ref =
    try Client.VDI.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      debug "Found no VDI with uuid = '%s' on the master, so creating one." my_uuid;
      Client.VDI.pool_introduce ~rpc ~session_id
        ~uuid:my_uuid
        ~name_label:vdi.API.vDI_name_label
        ~name_description:vdi.API.vDI_name_description
        ~sR:new_sr_ref
        ~_type:vdi.API.vDI_type
        ~sharable:vdi.API.vDI_sharable
        ~read_only:vdi.API.vDI_read_only
        ~other_config:vdi.API.vDI_other_config
        ~location:(Db.VDI.get_location ~__context ~self:vdi_ref)
        ~xenstore_data:vdi.API.vDI_xenstore_data
        ~sm_config:vdi.API.vDI_sm_config
        ~managed:vdi.API.vDI_managed
        ~virtual_size:vdi.API.vDI_virtual_size
        ~physical_utilisation:vdi.API.vDI_physical_utilisation
        ~metadata_of_pool:vdi.API.vDI_metadata_of_pool
        ~is_a_snapshot:vdi.API.vDI_is_a_snapshot
        ~snapshot_time:vdi.API.vDI_snapshot_time
        ~snapshot_of:vdi.API.vDI_snapshot_of in
  new_vdi_ref

let create_or_get_network_on_master __context rpc session_id (network_ref, network) : API.ref_network =
  let my_bridge = network.API.network_bridge in
  let is_physical = match network.API.network_PIFs with
    | [] -> false
    | hd :: _ -> Db.PIF.get_physical ~__context ~self:hd
  in
  let is_himn =
    (List.mem_assoc Xapi_globs.is_host_internal_management_network network.API.network_other_config) &&
    (List.assoc Xapi_globs.is_host_internal_management_network network.API.network_other_config = "true")
  in

  let new_network_ref =
    if is_physical || is_himn then
      (* Physical network or Host Internal Management Network:
         			 * try to join an existing network with the same bridge name, or create one.
         			 * This relies on the convention that physical PIFs with the same device name need to be connected.
         			 * Furthermore, there should be only one Host Internal Management Network in a pool. *)
      try
        let pool_networks = Client.Network.get_all_records ~rpc ~session_id in
        let net_ref, _ = List.find (fun (_, net) -> net.API.network_bridge = my_bridge) pool_networks in
        net_ref
      with _ ->
        debug "Found no network with bridge = '%s' on the master, so creating one." my_bridge;
        Client.Network.pool_introduce ~rpc ~session_id
          ~name_label:network.API.network_name_label
          ~name_description:network.API.network_name_description
          ~mTU:network.API.network_MTU
          ~other_config:network.API.network_other_config
          ~bridge:network.API.network_bridge
          ~managed:network.API.network_managed
    else begin
      debug "Recreating network '%s' as internal network." network.API.network_name_label;
      (* This call will generate a new 'xapi#' bridge name rather than keeping the
         			 * current, possibly colliding one. *)
      Client.Network.create ~rpc ~session_id
        ~name_label:network.API.network_name_label
        ~name_description:network.API.network_name_description
        ~mTU:network.API.network_MTU
        ~other_config:network.API.network_other_config
        ~bridge:network.API.network_bridge
        ~managed:network.API.network_managed
        ~tags:network.API.network_tags
    end
  in

  new_network_ref

let create_or_get_pif_on_master __context rpc session_id (pif_ref, pif) : API.ref_PIF =
  let my_uuid = pif.API.pIF_uuid in

  let my_host_ref = pif.API.pIF_host in
  let my_host = Db.Host.get_record ~__context ~self:my_host_ref in
  let new_host_ref = create_or_get_host_on_master __context rpc session_id (my_host_ref, my_host) in

  let new_network_ref =
    if pif.API.pIF_VLAN <> -1L then begin
      (* Get the remote management network for management VLAN PIF *)
      let remote_mgmt_pif = Client.Host.get_management_interface rpc session_id (get_master ~rpc ~session_id) in
      Client.PIF.get_network rpc session_id remote_mgmt_pif
    end
    else begin
      let my_network_ref = pif.API.pIF_network in
      let my_network = Db.Network.get_record ~__context ~self:my_network_ref in
      create_or_get_network_on_master __context rpc session_id (my_network_ref, my_network)
    end
  in

  let new_pif_ref =
    try Client.PIF.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      debug "Found no PIF with uuid = '%s' on the master, so creating one." my_uuid;
      Client.PIF.pool_introduce ~rpc ~session_id
        ~device:pif.API.pIF_device
        ~network:new_network_ref
        ~host:new_host_ref
        ~mAC:pif.API.pIF_MAC
        ~mTU:pif.API.pIF_MTU
        ~vLAN:pif.API.pIF_VLAN
        ~physical:pif.API.pIF_physical
        ~ip_configuration_mode:pif.API.pIF_ip_configuration_mode
        ~iP:pif.API.pIF_IP
        ~netmask:pif.API.pIF_netmask
        ~gateway:pif.API.pIF_gateway
        ~dNS:pif.API.pIF_DNS
        ~bond_slave_of:pif.API.pIF_bond_slave_of
        ~vLAN_master_of:pif.API.pIF_VLAN_master_of
        ~management:pif.API.pIF_management
        ~other_config:pif.API.pIF_other_config
        ~disallow_unplug:pif.API.pIF_disallow_unplug
        ~ipv6_configuration_mode:pif.API.pIF_ipv6_configuration_mode
        ~iPv6:pif.API.pIF_IPv6
        ~ipv6_gateway:pif.API.pIF_ipv6_gateway
        ~primary_address_type:pif.API.pIF_primary_address_type
        ~managed:pif.API.pIF_managed
        ~properties:pif.API.pIF_properties in

  new_pif_ref

let create_or_get_vlan_on_master __context rpc session_id (vlan_ref, vlan) : API.ref_VLAN =
  (* Create a VLAN PIF record only if it is a management PIF *)

  let my_host_ref = Db.PIF.get_host ~__context ~self:vlan.API.vLAN_tagged_PIF in
  let my_host = Db.Host.get_record ~__context ~self:my_host_ref in
  let new_host_ref = create_or_get_host_on_master __context rpc session_id (my_host_ref, my_host) in

  (* Create the untagged PIF record on Pool *)
  let untagged_pif_record = Db.PIF.get_record ~__context ~self:vlan.API.vLAN_untagged_PIF in
  let remote_untagged_pif = create_or_get_pif_on_master __context rpc session_id
    (vlan.API.vLAN_untagged_PIF, untagged_pif_record) in

  (* Get the remote tagged pif network *)
  let tagged_pif_network = Db.PIF.get_network ~__context ~self:vlan.API.vLAN_tagged_PIF in
  let tagged_pif_network_record = Db.Network.get_record ~__context ~self:tagged_pif_network in
  let remote_tagged_pif_network = create_or_get_network_on_master __context rpc session_id
    (tagged_pif_network, tagged_pif_network_record) in

  (* Get the new physical PIF ref on Pool for the joining Host *)
  let expr =
    Printf.sprintf "field \"network\"=\"%s\" and field \"host\"=\"%s\""
    (Ref.string_of remote_tagged_pif_network)
    (Ref.string_of new_host_ref)
  in
  let remote_physical_pif =
    match Client.PIF.get_all_records_where ~rpc ~session_id ~expr with
    | [] -> Ref.null
    | (pif, _) :: _ -> pif
  in

  let new_vlan_ref =
    try
      Client.VLAN.get_by_uuid ~rpc ~session_id ~uuid:vlan.API.vLAN_uuid
    with _ ->
      debug "Found no VLAN with uuid = '%s' on the master, so creating one." vlan.API.vLAN_uuid;
      Client.VLAN.pool_introduce ~rpc ~session_id
        ~tagged_PIF:remote_physical_pif
        ~untagged_PIF:remote_untagged_pif
        ~tag:vlan.API.vLAN_tag
        ~other_config:vlan.API.vLAN_other_config
  in
  new_vlan_ref

let create_or_get_pvs_site_on_master __context rpc session_id (pvs_site_ref, pvs_site) : API.ref_PVS_site =
  let my_pvs_uuid = pvs_site.API.pVS_site_PVS_uuid in
  let new_pvs_site_ref =
    let expr = "field \"PVS_uuid\"=\"" ^ my_pvs_uuid ^ "\"" in
    match Client.PVS_site.get_all_records_where ~rpc ~session_id ~expr with
    | [] ->
      debug "Found no PVS site with PVS_uuid = '%s' on the master, so creating one." my_pvs_uuid;
      let new_pvs_site = Client.PVS_site.introduce ~rpc ~session_id
          ~name_label:pvs_site.API.pVS_site_name_label
          ~name_description:pvs_site.API.pVS_site_name_description
          ~pVS_uuid:pvs_site.API.pVS_site_PVS_uuid in
      (* Update PVS servers *)
      let (_ : API.ref_PVS_server list) = List.map (fun pvs_server ->
          let pvs_record = Db.PVS_server.get_record ~__context ~self:pvs_server in
          Client.PVS_server.introduce ~rpc ~session_id
            ~addresses:pvs_record.API.pVS_server_addresses
            ~first_port:pvs_record.API.pVS_server_first_port
            ~last_port:pvs_record.API.pVS_server_last_port
            ~site:new_pvs_site
        ) (Db.PVS_site.get_servers ~__context ~self:pvs_site_ref) in
      new_pvs_site
    | (pvs_site, _) :: _ -> pvs_site
  in
  new_pvs_site_ref

let create_or_get_pvs_cache_storage_on_master __context rpc session_id (pcs_ref, pcs) : API.ref_PVS_cache_storage =
  let my_host_ref = pcs.API.pVS_cache_storage_host in
  let my_host = Db.Host.get_record ~__context ~self:my_host_ref in
  let new_host_ref = create_or_get_host_on_master __context rpc session_id (my_host_ref, my_host) in

  let my_sr_ref = pcs.API.pVS_cache_storage_SR in
  let my_sr = Db.SR.get_record ~__context ~self:my_sr_ref in
  let new_sr_ref = create_or_get_sr_on_master __context rpc session_id (my_sr_ref, my_sr) in

  let my_pvs_site_ref = pcs.API.pVS_cache_storage_site in
  let my_pvs_site = Db.PVS_site.get_record ~__context ~self:my_pvs_site_ref in
  let new_pvs_site_ref = create_or_get_pvs_site_on_master __context rpc session_id (my_pvs_site_ref, my_pvs_site) in

  let new_pvs_cache_storage_ref =
    let expr =
      Printf.sprintf "field \"site\"=\"%s\" and field \"host\"=\"%s\""
        (Ref.string_of new_pvs_site_ref)
        (Ref.string_of new_host_ref)
    in
    match Client.PVS_cache_storage.get_all_records_where ~rpc ~session_id ~expr with
    | [] ->
      Client.PVS_cache_storage.create ~rpc ~session_id
        ~host:new_host_ref
        ~sR:new_sr_ref
        ~site:new_pvs_site_ref
        ~size:pcs.API.pVS_cache_storage_size
    | (pvs_cache_storage, _) :: _ -> pvs_cache_storage
  in
  new_pvs_cache_storage_ref

let create_or_get_secret_on_master __context rpc session_id (secret_ref, secret) : API.ref_secret =
  let my_uuid = secret.API.secret_uuid in
  let my_value = secret.API.secret_value in
  let new_secret_ref =
    try Client.Secret.get_by_uuid ~rpc ~session_id ~uuid:my_uuid
    with _ ->
      debug "Found no secret with uuid = '%s' on master, so creating one." my_uuid;
      Client.Secret.introduce ~rpc ~session_id ~uuid:my_uuid ~value:my_value ~other_config:[]
  in
  new_secret_ref

let protect_exn f x =
  try Some (f x)
  with e ->
    debug "Ignoring exception: %s" (Printexc.to_string e);
    Debug.log_backtrace e (Backtrace.get e);
    None

(* Remark: the order in which we create the object in the distant database is not very important, as we have *)
(* an unique way to identify each object and thus we know if we need to create them or if it is already done *)
let update_non_vm_metadata ~__context ~rpc ~session_id =

  (* Update hosts *)
  let my_hosts = Db.Host.get_all_records ~__context in
  let (_ : API.ref_host option list) =
    List.map (protect_exn (create_or_get_host_on_master __context rpc session_id)) my_hosts in

  (* Update SRs *)
  let my_srs = Db.SR.get_all_records ~__context in
  let (_ : API.ref_SR option list) =
    List.map (protect_exn (create_or_get_sr_on_master __context rpc session_id)) my_srs in

  (* Update PBDs *)
  let my_pbds = Db.PBD.get_all_records ~__context in
  let (_ : API.ref_PBD option list) =
    List.map (protect_exn (create_or_get_pbd_on_master __context rpc session_id)) my_pbds in

  (* Update VDIs *)
  let my_vdis = Db.VDI.get_all_records ~__context in
  let (_ : API.ref_VDI option list) =
    List.map (protect_exn (create_or_get_vdi_on_master __context rpc session_id)) my_vdis in

  (* Update networks *)
  let my_networks = Db.Network.get_all_records ~__context in
  let (_ : API.ref_network option list) =
    List.map (protect_exn (create_or_get_network_on_master __context rpc session_id)) my_networks in

  (* update PIFs *)
  let my_pifs = Db.PIF.get_records_where ~__context ~expr:(
      Eq (Field "physical", Literal "true")
    ) in
  let (_ : API.ref_PIF option list) =
    List.map (protect_exn (create_or_get_pif_on_master __context rpc session_id)) my_pifs in

  (* update Management VLAN *)
  let mgmt_pif = Xapi_host.get_management_interface ~__context ~host:(Helpers.get_localhost ~__context) in
  let my_vlan = Db.PIF.get_VLAN_master_of ~__context ~self:mgmt_pif in
  if my_vlan <> Ref.null then
    let my_vlan_record = Db.VLAN.get_record ~__context ~self:my_vlan in
    let (_ : API.ref_VLAN option ) = protect_exn (create_or_get_vlan_on_master __context rpc session_id) (my_vlan, my_vlan_record) in

  (* update PVS sites *)
  let my_pvs_sites = Db.PVS_site.get_all_records ~__context in
  let (_ : API.ref_PVS_site option list) =
    List.map (protect_exn (create_or_get_pvs_site_on_master __context rpc session_id)) my_pvs_sites in

  (* update PVS_cache_storage *)
  let my_pvs_cache_storages = Db.PVS_cache_storage.get_all_records ~__context in
  let (_ : API.ref_PVS_cache_storage option list) =
{'fraction_non_alphanumeric': 0.07850436626116804, 'fraction_numerical': 0.0017035990106506486, 'mean_word_length': 3.604322816803207, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 157, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, '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': '3479449', 'n_tokens_mistral': 27184, 'n_tokens_neox': 26558, 'n_words': 6578}