<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.
 *)
(** Module that defines API functions for VDI objects
 * @group XenAPI functions
 *)
 
module D=Debug.Debugger(struct let name="xapi" end)
open D

open Pervasiveext
open Printf

(**************************************************************************************)
(* current/allowed operations checking                                                *)

(** Checks to see if an operation is valid in this state. Returns Some exception
    if not and None if everything is ok. *)
let check_operation_error ~__context ha_enabled record _ref' op = 
  let _ref = Ref.string_of _ref' in
  let current_ops = record.Db_actions.vDI_current_operations in

  let reset_on_boot = record.Db_actions.vDI_on_boot = `reset in

  (* Policy:
     1. any current_operation implies exclusivity; fail everything else
     2. if doing a VM start then assume the sharing check is done elsewhere
        (so VMs may share disks but our operations cannot)
     3. for other operations, fail if any VBD has currently-attached=true or any VBD 
        has a current_operation itself
     4. HA prevents you from deleting statefiles or metadata volumes
  *)
  if List.length current_ops > 0 
  then Some(Api_errors.other_operation_in_progress,["VDI"; _ref])
  else 
    (* check to see whether it's a local cd drive *)
    let sr = Db.VDI.get_SR ~__context ~self:_ref' in
	let sr_record = Db.SR.get_record_internal ~__context ~self:sr in
    let srtype = sr_record.Db_actions.sR_type in
    let is_tools_sr = Helpers.is_tools_sr ~__context ~sr in

    (* check to see whether VBDs exist which are using this VDI *)
    let vbds = Db.VDI.get_VBDs ~__context ~self:_ref' in
    let vbd_recs = List.map (fun self -> Db.VBD.get_record_internal ~__context ~self) vbds in

    (* Only a 'live' operation can be performed if there are active (even RO) devices *)
    let is_active v = v.Db_actions.vBD_currently_attached || v.Db_actions.vBD_reserved in
    (* VBD operations (plug/unplug) (which should be transient) cause us to serialise *)
    let has_current_operation v = v.Db_actions.vBD_current_operations <> [] in

    (* If the VBD is currently_attached then some operations can still be performed ie:
       VDI.clone (if the VM is suspended we have to have the 'allow_clone_suspended_vm'' flag)
       VDI.snapshot; VDI.resize_online; 'blocked' (CP-831) *)
    let operation_can_be_performed_live = match op with
	| `snapshot -> true
	| `resize_online -> true
	| `blocked -> true
	| `clone -> true
	| _ -> false in

    (* NB RO vs RW sharing checks are done in xapi_vbd.ml *)

	let sm_caps = Xapi_sr_operations.capabilities_of_sr sr_record in

    let any_vbd p = List.fold_left (||) false (List.map p vbd_recs) in
    if not operation_can_be_performed_live && (any_vbd is_active)
    then Some (Api_errors.vdi_in_use,[_ref; (Record_util.vdi_operation_to_string op)])
    else if any_vbd has_current_operation
    then Some (Api_errors.other_operation_in_progress, [ "VDI"; _ref ])
    else (
      match op with
	`destroy ->
	  if srtype = "udev" 
	  then Some (Api_errors.vdi_is_a_physical_device, [_ref])
	  else 
	    if is_tools_sr
	    then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	    else
	      if ha_enabled && List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	      then Some (Api_errors.ha_is_enabled, [])
	      else
		if not (List.mem Smint.Vdi_delete sm_caps)
		then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
		else None
      | `resize ->
	  if ha_enabled && List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	  then Some (Api_errors.ha_is_enabled, [])
	  else 
	    if not (List.mem Smint.Vdi_resize sm_caps)
	    then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	    else None
      | `update ->
	  if not (List.mem Smint.Vdi_update sm_caps) then
	    Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	  else None
      | `resize_online ->
	  if ha_enabled && List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	  then Some (Api_errors.ha_is_enabled, [])
	  else 
	    if not (List.mem Smint.Vdi_resize_online sm_caps)
	    then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	    else None
      | `generate_config ->
	  if not (List.mem Smint.Vdi_generate_config sm_caps) then
	    Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	  else None	    
	  | `snapshot when record.Db_actions.vDI_sharable ->
			Some (Api_errors.vdi_is_sharable, [ _ref ])
	  | `snapshot when reset_on_boot ->
		    Some (Api_errors.vdi_on_boot_mode_incompatable_with_operation, [])
	  | `copy ->
	  if List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	  then Some (Api_errors.operation_not_allowed, ["VDI containing HA statefile or redo log cannot be copied (check the VDI's allowed operations)."])
	  else None
	  | `clone ->
	      if not (List.mem Smint.Vdi_clone sm_caps)
	      then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	      else None
      | _ -> None
    )

let assert_operation_valid ~__context ~self ~(op:API.vdi_operations) = 
  let pool = Helpers.get_pool ~__context in
  let ha_enabled = Db.Pool.get_ha_enabled ~__context ~self:pool in

  let all = Db.VDI.get_record_internal ~__context ~self in
  match check_operation_error ~__context ha_enabled all self op with
      None -> ()
    | Some (a,b) -> raise (Api_errors.Server_error (a,b))

let update_allowed_operations ~__context ~self : unit =
  let pool = Helpers.get_pool ~__context in
  let ha_enabled = Db.Pool.get_ha_enabled ~__context ~self:pool in

  let all = Db.VDI.get_record_internal ~__context ~self in
  let allowed = 
    let check x = match check_operation_error ~__context ha_enabled all self x with None ->  [ x ] | _ -> [] in
	List.fold_left (fun accu op -> check op @ accu) []
		[ `snapshot; `copy; `clone; `destroy; `resize; `update; `generate_config; `resize_online ] in
  Db.VDI.set_allowed_operations ~__context ~self ~value:allowed

(** Someone is cancelling a task so remove it from the current_operations *)
let cancel_task ~__context ~self ~task_id = 
  let all = List.map fst (Db.VDI.get_current_operations ~__context ~self) in
  if List.mem task_id all then
    begin
      Db.VDI.remove_from_current_operations ~__context ~self ~key:task_id;
      update_allowed_operations ~__context ~self
    end

let cancel_tasks ~__context ~self ~all_tasks_in_db ~task_ids =
  let ops = Db.VDI.get_current_operations ~__context ~self in
  let set = (fun value -> Db.VDI.set_current_operations ~__context ~self ~value) in
  Helpers.cancel_tasks ~__context ~ops ~all_tasks_in_db ~task_ids ~set

(**************************************************************************************)

(** Helper function to create a new VDI record with all fields copied from
    an original, except ref and *_operations, UUID and others supplied as optional arguments.
    If a new UUID is not supplied, a fresh one is generated. 
    storage_lock defaults to false.
    Parent defaults to Ref.null.
 *)
(*let clone_record ~uuid ?name_label ?name_description ?sR ?virtual_size ?location
    ?physical_utilisation ?_type ?sharable ?read_only ?storage_lock ?other_config ?parent
    ?xenstore_data ?sm_config ~current_operations ~__context ~original () =
  let a = Db.VDI.get_record_internal ~__context ~self:original in
  let r = Ref.make () in
  Db.VDI.create ~__context ~ref:r 
    ~uuid:(Uuid.to_string uuid)
    ~name_label:(default a.Db_actions.vDI_name_label name_label)
    ~name_description:(default a.Db_actions.vDI_name_description name_description)
    ~allowed_operations:[] ~current_operations
    ~sR:(default a.Db_actions.vDI_SR sR)    
    ~virtual_size:(default a.Db_actions.vDI_virtual_size virtual_size)
    ~physical_utilisation:(default a.Db_actions.vDI_physical_utilisation physical_utilisation)
    ~_type:(default a.Db_actions.vDI_type _type)
    ~sharable:(default a.Db_actions.vDI_sharable sharable)
    ~read_only:(default a.Db_actions.vDI_read_only read_only)
    ~other_config:(default a.Db_actions.vDI_other_config other_config)
    ~storage_lock:(default false storage_lock)
    ~location:(default a.Db_actions.vDI_location location) ~managed:true ~missing:false
    ~xenstore_data:(default a.Db_actions.vDI_xenstore_data xenstore_data)
    ~sm_config:(default a.Db_actions.vDI_sm_config sm_config)
    ~parent:(default Ref.null parent);
  r*)

let require_uuid vdi_info = 
  match vdi_info.Smint.vdi_info_uuid with
  | Some uuid -> uuid
  | None -> failwith "SM backend failed to return <uuid> field" 

let newvdi ~__context ~sr newvdi =
	let open Storage_interface in
	let open Db_filter_types in
	let db_vdis = Db.VDI.get_records_where  ~__context ~expr:(Eq(Field "location", Literal newvdi.vdi)) in
	Xapi_sr.update_vdis ~__context ~sr db_vdis [ newvdi ];
	match Db.VDI.get_records_where  ~__context ~expr:(Eq(Field "location", Literal newvdi.vdi)) with
		| (vdi, _) :: _ -> vdi
		| [] -> failwith (Printf.sprintf "newvdi failed to create a VDI for %s" (string_of_vdi_info newvdi))

let default_vdi_info =
	let open Storage_interface in {
		vdi = "";
		sr = "";
		content_id = "";
		name_label = "";
		name_description = "";
		ty = "user";
		metadata_of_pool = "";
		is_a_snapshot = false;
		snapshot_time = Date.to_string Date.never;
		snapshot_of = "";
		read_only = false;
		virtual_size = 0L;
		physical_utilisation = 0L
	}

let create ~__context ~name_label ~name_description
        ~sR ~virtual_size ~_type
        ~sharable ~read_only ~other_config ~xenstore_data ~sm_config ~tags =
    Sm.assert_pbd_is_plugged ~__context ~sr:sR;

	(* XXX: unify with record_util.vdi_type_to_string *)
    let vdi_type = match _type with
        | `crashdump -> "crashdump"
        | `ephemeral -> "ephemeral"
        | `ha_statefile -> "ha_statefile"
        | `metadata -> "metadata"
        | `redo_log -> "redo_log"
        | `suspend -> "suspend"
        | `system -> "system"
        | `user -> "user" in

    let open Storage_access in
    let task = Context.get_task_id __context in
    let open Storage_interface in
	let vdi_info = {
		default_vdi_info with
		name_label = name_label;
		name_description = name_description;
		ty = vdi_type;
		read_only = read_only;
		virtual_size = virtual_size;
	} in
	let module C = Client(struct let rpc = rpc end) in
	let vi = transform_storage_exn
		(fun () -> C.VDI.create ~dbg:(Ref.string_of task) ~sr:(Db.SR.get_uuid ~__context ~self:sR)
			~vdi_info ~params:sm_config) in
	if virtual_size < vi.virtual_size
	then info "sr:%s vdi:%s requested virtual size %Ld < actual virtual size %Ld" (Ref.string_of sR) vi.vdi virtual_size vi.virtual_size;
	newvdi ~__context ~sr:sR vi



(* Make the database record only *)
let introduce_dbonly  ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config  ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of =
  (* Verify that the location field is unique in this SR *)
  List.iter
    (fun vdi ->
       if Db.VDI.get_location ~__context ~self:vdi = location
       then raise (Api_errors.Server_error (Api_errors.location_not_unique, [ Ref.string_of sR; location ]))
    ) (Db.SR.get_VDIs ~__context ~self:sR);
  (* Verify the UUID field looks like a UUID *)
  begin
    try
      Scanf.sscanf uuid "%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x"
	(fun _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ -> ())
    with _ -> raise (Api_errors.Server_error (Api_errors.uuid_invalid, [ "VDI"; uuid ]))
  end;
  let ref = Ref.make() in
  debug "VDI.introduce read_only = %b" read_only;
  Db.VDI.create ~__context ~ref ~uuid:uuid
    ~name_label ~name_description 
    ~current_operations:[] ~allowed_operations:[]
    ~is_a_snapshot ~snapshot_of ~snapshot_time
    ~sR ~virtual_size
    ~physical_utilisation ~_type
    ~sharable ~read_only
    ~xenstore_data ~sm_config
    ~other_config ~storage_lock:false ~location ~managed ~missing:false ~parent:Ref.null ~tags:[]
    ~on_boot:`persist ~allow_caching:false
    ~metadata_of_pool ~metadata_latest:false;
  ref

let internal_db_introduce ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of =
  debug "{pool,db}_introduce uuid=%s name_label=%s" uuid name_label;
  let ref = introduce_dbonly ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config  ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of in
  update_allowed_operations ~__context ~self:ref;
  ref
  
let pool_introduce = internal_db_introduce
let db_introduce = internal_db_introduce

let db_forget ~__context ~vdi = 
  debug "db_forget uuid=%s" (Db.VDI.get_uuid ~__context ~self:vdi);
  Db.VDI.destroy ~__context ~self:vdi

let introduce ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of =
  debug "introduce uuid=%s name_label=%s sm_config=[ %s ]" uuid name_label (String.concat "; " (List.map (fun (k, v) -> k ^ " = " ^ v) sm_config));  
  Sm.assert_pbd_is_plugged ~__context ~sr:sR;
  let vdi_info = 
    Sm.call_sm_functions ~__context ~sR
      (fun device_config sr_type ->
	 Sm.vdi_introduce device_config sr_type sR uuid sm_config location)
  in
  let uuid = require_uuid vdi_info in
  let ref = Db.VDI.get_by_uuid ~__context ~uuid in

  (* Set the fields which belong to the higher-level API: *)
  Db.VDI.set_other_config ~__context ~self:ref ~value:other_config;
  Db.VDI.set_xenstore_data ~__context ~self:ref ~value:xenstore_data;
  Db.VDI.set_sharable ~__context ~self:ref ~value:sharable;
  Db.VDI.set_type ~__context ~self:ref ~value:_type;

  (* CA-13140: The udev SR has already filled in the name and description, so
     unless we've been called with overrides, don't set these fields *)
  if name_label <> "" then
    Db.VDI.set_name_label ~__context ~self:ref ~value:name_label;
  if name_description <> "" then
    Db.VDI.set_name_description ~__context ~self:ref ~value:name_description;

  update_allowed_operations ~__context ~self:ref;
  ref

let update ~__context ~vdi = 
  debug "update ref=%s location=%s" (Ref.string_of vdi) (Db.VDI.get_location ~__context ~self:vdi);
  let sR = Db.VDI.get_SR ~__context ~self:vdi in
  Sm.assert_pbd_is_plugged ~__context ~sr:sR;
  Sm.call_sm_functions ~__context ~sR
    (fun device_config sr_type ->
       Sm.vdi_update device_config sr_type sR vdi)

let forget ~__context ~vdi = Db.VDI.destroy ~__context ~self:vdi

open Client

(* driver_params is the storage-backend-specific parameters that are used to drive the
   snapshot operation (e.g. vmhint for NetAPP)
*)
let snapshot_and_clone call_f ~__context ~vdi ~driver_params =
	let sR = Db.VDI.get_SR ~__context ~self:vdi in
  Sm.assert_pbd_is_plugged ~__context ~sr:sR;
  Xapi_vdi_helpers.assert_managed ~__context ~vdi;
  let a = Db.VDI.get_record_internal ~__context ~self:vdi in

  let call_snapshot () = 
	  let open Storage_access in
	  let task = Context.get_task_id __context in	
	  let open Storage_interface in
	  let vdi_info = {
		  default_vdi_info with
			  name_label = a.Db_actions.vDI_name_label;
			  name_description = a.Db_actions.vDI_name_description
	  } in
	  let sr' = Db.SR.get_uuid ~__context ~self:sR in
	  let vdi' = Db.VDI.get_location ~__context ~self:vdi in
	  (* We don't use transform_storage_exn because of the clone/copy fallback below *)
	  let new_vdi = call_f ~dbg:(Ref.string_of task) ~sr:sr'
		  ~vdi:vdi' ~vdi_info  ~params:driver_params in
	  newvdi ~__context ~sr:sR new_vdi
  in

  (* While we don't have blkback support for pause/unpause we only do this
     for .vhd-based backends. *)
  let newvdi = call_snapshot () in

  (* Copy across the metadata which we control *)
  Db.VDI.set_name_label ~__context ~self:newvdi ~value:a.Db_actions.vDI_name_label;
  Db.VDI.set_name_description ~__context ~self:newvdi ~value:a.Db_actions.vDI_name_description;
  Db.VDI.set_type ~__context ~self:newvdi ~value:a.Db_actions.vDI_type;
  Db.VDI.set_sharable ~__context ~self:newvdi ~value:a.Db_actions.vDI_sharable;
  Db.VDI.set_other_config ~__context ~self:newvdi ~value:a.Db_actions.vDI_other_config;
  Db.VDI.set_xenstore_data ~__context ~self:newvdi ~value:a.Db_actions.vDI_xenstore_data;
  Db.VDI.set_on_boot ~__context ~self:newvdi ~value:a.Db_actions.vDI_on_boot;
  Db.VDI.set_allow_caching ~__context ~self:newvdi ~value:a.Db_actions.vDI_allow_caching;
  newvdi

let snapshot ~__context ~vdi ~driver_params =
	let module C = Storage_interface.Client(struct let rpc = Storage_access.rpc end) in
	let newvdi = Storage_access.transform_storage_exn
		(fun () ->
			try
				snapshot_and_clone C.VDI.snapshot ~__context ~vdi ~driver_params
			with Storage_interface.Unimplemented _ ->
				(* CA-28598 *)
				debug "Backend reported not implemented despite it offering the capability; assuming this is an LVHD upgrade issue";
				raise (Api_errors.Server_error(Api_errors.sr_requires_upgrade, [ Ref.string_of (Db.VDI.get_SR ~__context ~self:vdi) ]))
		) in
	(* Record the fact this is a snapshot *)
 
	(*(try Db.VDI.remove_from_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_of with _ -> ());
	  (try Db.VDI.remove_from_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_time with _ -> ());
	  Db.VDI.add_to_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_of ~value:a.Db_actions.vDI_uuid;
	  Db.VDI.add_to_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_time ~value:(Date.to_string (Date.of_float (Unix.gettimeofday ())));*)
	Db.VDI.set_is_a_snapshot ~__context ~self:newvdi ~value:true;
	Db.VDI.set_snapshot_of ~__context ~self:newvdi ~value:vdi;
	Db.VDI.set_snapshot_time ~__context ~self:newvdi ~value:(Date.of_float (Unix.gettimeofday ()));

	update_allowed_operations ~__context ~self:newvdi;
	update ~__context ~vdi:newvdi;
	newvdi

let destroy ~__context ~self =
	let sr = Db.VDI.get_SR ~__context ~self in
	let location = Db.VDI.get_location ~__context ~self in
  Sm.assert_pbd_is_plugged ~__context ~sr;
  Xapi_vdi_helpers.assert_managed ~__context ~vdi:self;

  let vbds = Db.VDI.get_VBDs ~__context ~self in
  let attached_vbds = List.filter 
    (fun vbd->
       let r = Db.VBD.get_record_internal ~__context ~self:vbd in
       r.Db_actions.vBD_currently_attached || r.Db_actions.vBD_reserved) vbds in
    if attached_vbds<>[] then
      raise (Api_errors.Server_error (Api_errors.vdi_in_use, [(Ref.string_of self); "destroy" ]))
    else
      begin
          let open Storage_access in
          let open Storage_interface in
          let task = Context.get_task_id __context in
          let module C = Client(struct let rpc = rpc end) in
		  transform_storage_exn
			  (fun () ->
				  C.VDI.destroy ~dbg:(Ref.string_of task) ~sr:(Db.SR.get_uuid ~__context ~self:sr) ~vdi:location
			  );

	(* destroy all the VBDs now rather than wait for the GC thread. This helps
	   prevent transient glitches but doesn't totally prevent races. *)
	List.iter (fun vbd ->
		     Helpers.log_exn_continue (Printf.sprintf "destroying VBD: %s" (Ref.string_of vbd))
		       (fun vbd -> Db.VBD.destroy ~__context ~self:vbd) vbd) vbds;
	(* Db.VDI.destroy ~__context ~self *)
      end
<commit_msg>[sm] don't rely on the SM backend deleting the VDI record from the database<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.
 *)
(** Module that defines API functions for VDI objects
 * @group XenAPI functions
 *)
 
module D=Debug.Debugger(struct let name="xapi" end)
open D

open Pervasiveext
open Printf

(**************************************************************************************)
(* current/allowed operations checking                                                *)

(** Checks to see if an operation is valid in this state. Returns Some exception
    if not and None if everything is ok. *)
let check_operation_error ~__context ha_enabled record _ref' op = 
  let _ref = Ref.string_of _ref' in
  let current_ops = record.Db_actions.vDI_current_operations in

  let reset_on_boot = record.Db_actions.vDI_on_boot = `reset in

  (* Policy:
     1. any current_operation implies exclusivity; fail everything else
     2. if doing a VM start then assume the sharing check is done elsewhere
        (so VMs may share disks but our operations cannot)
     3. for other operations, fail if any VBD has currently-attached=true or any VBD 
        has a current_operation itself
     4. HA prevents you from deleting statefiles or metadata volumes
  *)
  if List.length current_ops > 0 
  then Some(Api_errors.other_operation_in_progress,["VDI"; _ref])
  else 
    (* check to see whether it's a local cd drive *)
    let sr = Db.VDI.get_SR ~__context ~self:_ref' in
	let sr_record = Db.SR.get_record_internal ~__context ~self:sr in
    let srtype = sr_record.Db_actions.sR_type in
    let is_tools_sr = Helpers.is_tools_sr ~__context ~sr in

    (* check to see whether VBDs exist which are using this VDI *)
    let vbds = Db.VDI.get_VBDs ~__context ~self:_ref' in
    let vbd_recs = List.map (fun self -> Db.VBD.get_record_internal ~__context ~self) vbds in

    (* Only a 'live' operation can be performed if there are active (even RO) devices *)
    let is_active v = v.Db_actions.vBD_currently_attached || v.Db_actions.vBD_reserved in
    (* VBD operations (plug/unplug) (which should be transient) cause us to serialise *)
    let has_current_operation v = v.Db_actions.vBD_current_operations <> [] in

    (* If the VBD is currently_attached then some operations can still be performed ie:
       VDI.clone (if the VM is suspended we have to have the 'allow_clone_suspended_vm'' flag)
       VDI.snapshot; VDI.resize_online; 'blocked' (CP-831) *)
    let operation_can_be_performed_live = match op with
	| `snapshot -> true
	| `resize_online -> true
	| `blocked -> true
	| `clone -> true
	| _ -> false in

    (* NB RO vs RW sharing checks are done in xapi_vbd.ml *)

	let sm_caps = Xapi_sr_operations.capabilities_of_sr sr_record in

    let any_vbd p = List.fold_left (||) false (List.map p vbd_recs) in
    if not operation_can_be_performed_live && (any_vbd is_active)
    then Some (Api_errors.vdi_in_use,[_ref; (Record_util.vdi_operation_to_string op)])
    else if any_vbd has_current_operation
    then Some (Api_errors.other_operation_in_progress, [ "VDI"; _ref ])
    else (
      match op with
	`destroy ->
	  if srtype = "udev" 
	  then Some (Api_errors.vdi_is_a_physical_device, [_ref])
	  else 
	    if is_tools_sr
	    then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	    else
	      if ha_enabled && List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	      then Some (Api_errors.ha_is_enabled, [])
	      else
		if not (List.mem Smint.Vdi_delete sm_caps)
		then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
		else None
      | `resize ->
	  if ha_enabled && List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	  then Some (Api_errors.ha_is_enabled, [])
	  else 
	    if not (List.mem Smint.Vdi_resize sm_caps)
	    then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	    else None
      | `update ->
	  if not (List.mem Smint.Vdi_update sm_caps) then
	    Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	  else None
      | `resize_online ->
	  if ha_enabled && List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	  then Some (Api_errors.ha_is_enabled, [])
	  else 
	    if not (List.mem Smint.Vdi_resize_online sm_caps)
	    then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	    else None
      | `generate_config ->
	  if not (List.mem Smint.Vdi_generate_config sm_caps) then
	    Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	  else None	    
	  | `snapshot when record.Db_actions.vDI_sharable ->
			Some (Api_errors.vdi_is_sharable, [ _ref ])
	  | `snapshot when reset_on_boot ->
		    Some (Api_errors.vdi_on_boot_mode_incompatable_with_operation, [])
	  | `copy ->
	  if List.mem record.Db_actions.vDI_type [ `ha_statefile; `redo_log ]
	  then Some (Api_errors.operation_not_allowed, ["VDI containing HA statefile or redo log cannot be copied (check the VDI's allowed operations)."])
	  else None
	  | `clone ->
	      if not (List.mem Smint.Vdi_clone sm_caps)
	      then Some (Api_errors.sr_operation_not_supported, [Ref.string_of sr])
	      else None
      | _ -> None
    )

let assert_operation_valid ~__context ~self ~(op:API.vdi_operations) = 
  let pool = Helpers.get_pool ~__context in
  let ha_enabled = Db.Pool.get_ha_enabled ~__context ~self:pool in

  let all = Db.VDI.get_record_internal ~__context ~self in
  match check_operation_error ~__context ha_enabled all self op with
      None -> ()
    | Some (a,b) -> raise (Api_errors.Server_error (a,b))

let update_allowed_operations ~__context ~self : unit =
  let pool = Helpers.get_pool ~__context in
  let ha_enabled = Db.Pool.get_ha_enabled ~__context ~self:pool in

  let all = Db.VDI.get_record_internal ~__context ~self in
  let allowed = 
    let check x = match check_operation_error ~__context ha_enabled all self x with None ->  [ x ] | _ -> [] in
	List.fold_left (fun accu op -> check op @ accu) []
		[ `snapshot; `copy; `clone; `destroy; `resize; `update; `generate_config; `resize_online ] in
  Db.VDI.set_allowed_operations ~__context ~self ~value:allowed

(** Someone is cancelling a task so remove it from the current_operations *)
let cancel_task ~__context ~self ~task_id = 
  let all = List.map fst (Db.VDI.get_current_operations ~__context ~self) in
  if List.mem task_id all then
    begin
      Db.VDI.remove_from_current_operations ~__context ~self ~key:task_id;
      update_allowed_operations ~__context ~self
    end

let cancel_tasks ~__context ~self ~all_tasks_in_db ~task_ids =
  let ops = Db.VDI.get_current_operations ~__context ~self in
  let set = (fun value -> Db.VDI.set_current_operations ~__context ~self ~value) in
  Helpers.cancel_tasks ~__context ~ops ~all_tasks_in_db ~task_ids ~set

(**************************************************************************************)

(** Helper function to create a new VDI record with all fields copied from
    an original, except ref and *_operations, UUID and others supplied as optional arguments.
    If a new UUID is not supplied, a fresh one is generated. 
    storage_lock defaults to false.
    Parent defaults to Ref.null.
 *)
(*let clone_record ~uuid ?name_label ?name_description ?sR ?virtual_size ?location
    ?physical_utilisation ?_type ?sharable ?read_only ?storage_lock ?other_config ?parent
    ?xenstore_data ?sm_config ~current_operations ~__context ~original () =
  let a = Db.VDI.get_record_internal ~__context ~self:original in
  let r = Ref.make () in
  Db.VDI.create ~__context ~ref:r 
    ~uuid:(Uuid.to_string uuid)
    ~name_label:(default a.Db_actions.vDI_name_label name_label)
    ~name_description:(default a.Db_actions.vDI_name_description name_description)
    ~allowed_operations:[] ~current_operations
    ~sR:(default a.Db_actions.vDI_SR sR)    
    ~virtual_size:(default a.Db_actions.vDI_virtual_size virtual_size)
    ~physical_utilisation:(default a.Db_actions.vDI_physical_utilisation physical_utilisation)
    ~_type:(default a.Db_actions.vDI_type _type)
    ~sharable:(default a.Db_actions.vDI_sharable sharable)
    ~read_only:(default a.Db_actions.vDI_read_only read_only)
    ~other_config:(default a.Db_actions.vDI_other_config other_config)
    ~storage_lock:(default false storage_lock)
    ~location:(default a.Db_actions.vDI_location location) ~managed:true ~missing:false
    ~xenstore_data:(default a.Db_actions.vDI_xenstore_data xenstore_data)
    ~sm_config:(default a.Db_actions.vDI_sm_config sm_config)
    ~parent:(default Ref.null parent);
  r*)

let require_uuid vdi_info = 
  match vdi_info.Smint.vdi_info_uuid with
  | Some uuid -> uuid
  | None -> failwith "SM backend failed to return <uuid> field" 

let newvdi ~__context ~sr newvdi =
	let open Storage_interface in
	let open Db_filter_types in
	let db_vdis = Db.VDI.get_records_where  ~__context ~expr:(Eq(Field "location", Literal newvdi.vdi)) in
	Xapi_sr.update_vdis ~__context ~sr db_vdis [ newvdi ];
	match Db.VDI.get_records_where  ~__context ~expr:(Eq(Field "location", Literal newvdi.vdi)) with
		| (vdi, _) :: _ -> vdi
		| [] -> failwith (Printf.sprintf "newvdi failed to create a VDI for %s" (string_of_vdi_info newvdi))

let default_vdi_info =
	let open Storage_interface in {
		vdi = "";
		sr = "";
		content_id = "";
		name_label = "";
		name_description = "";
		ty = "user";
		metadata_of_pool = "";
		is_a_snapshot = false;
		snapshot_time = Date.to_string Date.never;
		snapshot_of = "";
		read_only = false;
		virtual_size = 0L;
		physical_utilisation = 0L
	}

let create ~__context ~name_label ~name_description
        ~sR ~virtual_size ~_type
        ~sharable ~read_only ~other_config ~xenstore_data ~sm_config ~tags =
    Sm.assert_pbd_is_plugged ~__context ~sr:sR;

	(* XXX: unify with record_util.vdi_type_to_string *)
    let vdi_type = match _type with
        | `crashdump -> "crashdump"
        | `ephemeral -> "ephemeral"
        | `ha_statefile -> "ha_statefile"
        | `metadata -> "metadata"
        | `redo_log -> "redo_log"
        | `suspend -> "suspend"
        | `system -> "system"
        | `user -> "user" in

    let open Storage_access in
    let task = Context.get_task_id __context in
    let open Storage_interface in
	let vdi_info = {
		default_vdi_info with
		name_label = name_label;
		name_description = name_description;
		ty = vdi_type;
		read_only = read_only;
		virtual_size = virtual_size;
	} in
	let module C = Client(struct let rpc = rpc end) in
	let vi = transform_storage_exn
		(fun () -> C.VDI.create ~dbg:(Ref.string_of task) ~sr:(Db.SR.get_uuid ~__context ~self:sR)
			~vdi_info ~params:sm_config) in
	if virtual_size < vi.virtual_size
	then info "sr:%s vdi:%s requested virtual size %Ld < actual virtual size %Ld" (Ref.string_of sR) vi.vdi virtual_size vi.virtual_size;
	newvdi ~__context ~sr:sR vi



(* Make the database record only *)
let introduce_dbonly  ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config  ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of =
  (* Verify that the location field is unique in this SR *)
  List.iter
    (fun vdi ->
       if Db.VDI.get_location ~__context ~self:vdi = location
       then raise (Api_errors.Server_error (Api_errors.location_not_unique, [ Ref.string_of sR; location ]))
    ) (Db.SR.get_VDIs ~__context ~self:sR);
  (* Verify the UUID field looks like a UUID *)
  begin
    try
      Scanf.sscanf uuid "%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x"
	(fun _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ -> ())
    with _ -> raise (Api_errors.Server_error (Api_errors.uuid_invalid, [ "VDI"; uuid ]))
  end;
  let ref = Ref.make() in
  debug "VDI.introduce read_only = %b" read_only;
  Db.VDI.create ~__context ~ref ~uuid:uuid
    ~name_label ~name_description 
    ~current_operations:[] ~allowed_operations:[]
    ~is_a_snapshot ~snapshot_of ~snapshot_time
    ~sR ~virtual_size
    ~physical_utilisation ~_type
    ~sharable ~read_only
    ~xenstore_data ~sm_config
    ~other_config ~storage_lock:false ~location ~managed ~missing:false ~parent:Ref.null ~tags:[]
    ~on_boot:`persist ~allow_caching:false
    ~metadata_of_pool ~metadata_latest:false;
  ref

let internal_db_introduce ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of =
  debug "{pool,db}_introduce uuid=%s name_label=%s" uuid name_label;
  let ref = introduce_dbonly ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config  ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of in
  update_allowed_operations ~__context ~self:ref;
  ref
  
let pool_introduce = internal_db_introduce
let db_introduce = internal_db_introduce

let db_forget ~__context ~vdi = 
  debug "db_forget uuid=%s" (Db.VDI.get_uuid ~__context ~self:vdi);
  Db.VDI.destroy ~__context ~self:vdi

let introduce ~__context ~uuid ~name_label ~name_description ~sR ~_type ~sharable ~read_only ~other_config ~location ~xenstore_data ~sm_config ~managed ~virtual_size ~physical_utilisation ~metadata_of_pool ~is_a_snapshot ~snapshot_time ~snapshot_of =
  debug "introduce uuid=%s name_label=%s sm_config=[ %s ]" uuid name_label (String.concat "; " (List.map (fun (k, v) -> k ^ " = " ^ v) sm_config));  
  Sm.assert_pbd_is_plugged ~__context ~sr:sR;
  let vdi_info = 
    Sm.call_sm_functions ~__context ~sR
      (fun device_config sr_type ->
	 Sm.vdi_introduce device_config sr_type sR uuid sm_config location)
  in
  let uuid = require_uuid vdi_info in
  let ref = Db.VDI.get_by_uuid ~__context ~uuid in

  (* Set the fields which belong to the higher-level API: *)
  Db.VDI.set_other_config ~__context ~self:ref ~value:other_config;
  Db.VDI.set_xenstore_data ~__context ~self:ref ~value:xenstore_data;
  Db.VDI.set_sharable ~__context ~self:ref ~value:sharable;
  Db.VDI.set_type ~__context ~self:ref ~value:_type;

  (* CA-13140: The udev SR has already filled in the name and description, so
     unless we've been called with overrides, don't set these fields *)
  if name_label <> "" then
    Db.VDI.set_name_label ~__context ~self:ref ~value:name_label;
  if name_description <> "" then
    Db.VDI.set_name_description ~__context ~self:ref ~value:name_description;

  update_allowed_operations ~__context ~self:ref;
  ref

let update ~__context ~vdi = 
  debug "update ref=%s location=%s" (Ref.string_of vdi) (Db.VDI.get_location ~__context ~self:vdi);
  let sR = Db.VDI.get_SR ~__context ~self:vdi in
  Sm.assert_pbd_is_plugged ~__context ~sr:sR;
  Sm.call_sm_functions ~__context ~sR
    (fun device_config sr_type ->
       Sm.vdi_update device_config sr_type sR vdi)

let forget ~__context ~vdi = Db.VDI.destroy ~__context ~self:vdi

open Client

(* driver_params is the storage-backend-specific parameters that are used to drive the
   snapshot operation (e.g. vmhint for NetAPP)
*)
let snapshot_and_clone call_f ~__context ~vdi ~driver_params =
	let sR = Db.VDI.get_SR ~__context ~self:vdi in
  Sm.assert_pbd_is_plugged ~__context ~sr:sR;
  Xapi_vdi_helpers.assert_managed ~__context ~vdi;
  let a = Db.VDI.get_record_internal ~__context ~self:vdi in

  let call_snapshot () = 
	  let open Storage_access in
	  let task = Context.get_task_id __context in	
	  let open Storage_interface in
	  let vdi_info = {
		  default_vdi_info with
			  name_label = a.Db_actions.vDI_name_label;
			  name_description = a.Db_actions.vDI_name_description
	  } in
	  let sr' = Db.SR.get_uuid ~__context ~self:sR in
	  let vdi' = Db.VDI.get_location ~__context ~self:vdi in
	  (* We don't use transform_storage_exn because of the clone/copy fallback below *)
	  let new_vdi = call_f ~dbg:(Ref.string_of task) ~sr:sr'
		  ~vdi:vdi' ~vdi_info  ~params:driver_params in
	  newvdi ~__context ~sr:sR new_vdi
  in

  (* While we don't have blkback support for pause/unpause we only do this
     for .vhd-based backends. *)
  let newvdi = call_snapshot () in

  (* Copy across the metadata which we control *)
  Db.VDI.set_name_label ~__context ~self:newvdi ~value:a.Db_actions.vDI_name_label;
  Db.VDI.set_name_description ~__context ~self:newvdi ~value:a.Db_actions.vDI_name_description;
  Db.VDI.set_type ~__context ~self:newvdi ~value:a.Db_actions.vDI_type;
  Db.VDI.set_sharable ~__context ~self:newvdi ~value:a.Db_actions.vDI_sharable;
  Db.VDI.set_other_config ~__context ~self:newvdi ~value:a.Db_actions.vDI_other_config;
  Db.VDI.set_xenstore_data ~__context ~self:newvdi ~value:a.Db_actions.vDI_xenstore_data;
  Db.VDI.set_on_boot ~__context ~self:newvdi ~value:a.Db_actions.vDI_on_boot;
  Db.VDI.set_allow_caching ~__context ~self:newvdi ~value:a.Db_actions.vDI_allow_caching;
  newvdi

let snapshot ~__context ~vdi ~driver_params =
	let module C = Storage_interface.Client(struct let rpc = Storage_access.rpc end) in
	let newvdi = Storage_access.transform_storage_exn
		(fun () ->
			try
				snapshot_and_clone C.VDI.snapshot ~__context ~vdi ~driver_params
			with Storage_interface.Unimplemented _ ->
				(* CA-28598 *)
				debug "Backend reported not implemented despite it offering the capability; assuming this is an LVHD upgrade issue";
				raise (Api_errors.Server_error(Api_errors.sr_requires_upgrade, [ Ref.string_of (Db.VDI.get_SR ~__context ~self:vdi) ]))
		) in
	(* Record the fact this is a snapshot *)
 
	(*(try Db.VDI.remove_from_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_of with _ -> ());
	  (try Db.VDI.remove_from_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_time with _ -> ());
	  Db.VDI.add_to_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_of ~value:a.Db_actions.vDI_uuid;
	  Db.VDI.add_to_other_config ~__context ~self:newvdi ~key:Xapi_globs.snapshot_time ~value:(Date.to_string (Date.of_float (Unix.gettimeofday ())));*)
	Db.VDI.set_is_a_snapshot ~__context ~self:newvdi ~value:true;
	Db.VDI.set_snapshot_of ~__context ~self:newvdi ~value:vdi;
	Db.VDI.set_snapshot_time ~__context ~self:newvdi ~value:(Date.of_float (Unix.gettimeofday ()));

	update_allowed_operations ~__context ~self:newvdi;
	update ~__context ~vdi:newvdi;
	newvdi

let destroy ~__context ~self =
	let sr = Db.VDI.get_SR ~__context ~self in
	let location = Db.VDI.get_location ~__context ~self in
  Sm.assert_pbd_is_plugged ~__context ~sr;
  Xapi_vdi_helpers.assert_managed ~__context ~vdi:self;

  let vbds = Db.VDI.get_VBDs ~__context ~self in
  let attached_vbds = List.filter 
    (fun vbd->
       let r = Db.VBD.get_record_internal ~__context ~self:vbd in
       r.Db_actions.vBD_currently_attached || r.Db_actions.vBD_reserved) vbds in
    if attached_vbds<>[] then
      raise (Api_errors.Server_error (Api_errors.vdi_in_use, [(Ref.string_of self); "destroy" ]))
    else
      begin
          let open Storage_access in
          let open Storage_interface in
          let task = Context.get_task_id __context in
          let module C = Client(struct let rpc = rpc end) in
		  transform_storage_exn
			  (fun () ->
				  C.VDI.destroy ~dbg:(Ref.string_of task) ~sr:(Db.SR.get_uuid ~__context ~self:sr) ~vdi:location
			  );
		  if Db.is_valid_ref __context self
		  then Db.VDI.destroy ~__context ~self;

	(* destroy all the VBDs now rather than wait for the GC thread. This helps
	   prevent transient glitches but doesn't totally prevent races. *)
	List.iter (fun vbd ->
		     Helpers.log_exn_continue (Printf.sprintf "destroying VBD: %s" (Ref.string_of vbd))
		       (fun vbd -> Db.VBD.destroy ~__context ~self:vbd) vbd) vbds;
	(* Db.VDI.destroy ~__context ~self *)
      end

{'fraction_non_alphanumeric': 0.10441321499013807, 'fraction_numerical': 0.003057199211045365, 'mean_word_length': 4.157151938970121, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 113, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '6163649', 'n_tokens_mistral': 14624, 'n_tokens_neox': 14006, 'n_words': 3669}