CA-388210: use unique datapaths for concurrent VDI copies (#5920)

Testing on a box shows that the new unique arguments for 'domain' work,
but opening as a draft PR, because this needs wider testing in XenRT.

Author: edwintorok

ocaml/xapi-storage-script/lib.ml

@@ -131,6 +131,7 @@ module Process = struct
 
     type t = {
         exit_status: (unit, exit_or_signal) Result.t
+      ; pid: int
       ; stdout: string
       ; stderr: string
     }
@@ -176,6 +177,7 @@ module Process = struct
   let run ~env ~prog ~args ~input =
     let ( let@ ) f x = f x in
     let@ p = with_process ~env ~prog ~args in
+    let pid = p#pid in
     let sender = send p#stdin input in
     let receiver_out = receive p#stdout in
     let receiver_err = receive p#stderr in
@@ -185,7 +187,7 @@ module Process = struct
         Lwt.both sender receiver >>= fun ((), (stdout, stderr)) ->
         p#status >>= fun status ->
         let exit_status = Output.exit_or_signal_of_unix status in
-        Lwt.return {Output.exit_status; stdout; stderr}
+        Lwt.return {Output.exit_status; pid; stdout; stderr}
       )
       (function
         | Lwt.Canceled as exn ->

ocaml/xapi-storage-script/lib.mli

@@ -65,6 +65,7 @@ module Process : sig
 
     type t = {
         exit_status: (unit, exit_or_signal) result
+      ; pid: int
       ; stdout: string
       ; stderr: string
     }
@@ -78,7 +79,7 @@ module Process : sig
     -> Output.t Lwt.t
   (** Runs a cli program, writes [input] into its stdin, then closing the fd,
       and finally waits for the program to finish and returns the exit status,
-      its stdout and stderr. *)
+      the pid, and its stdout and stderr. *)
 end
 
 module DirWatcher : sig

ocaml/xapi-storage-script/main.ml

@@ -66,6 +66,23 @@ let backend_backtrace_error name args backtrace =
 let missing_uri () =
   backend_error "MISSING_URI" ["Please include a URI in the device-config"]
 
+(** return a unique 'domain' string for Dom0, so that we can plug disks
+  multiple times (e.g. for copy).
+
+  XAPI should give us a unique 'dp' (datapath) string, e.g. a UUID for storage migration,
+  or vbd/domid/device.
+  For regular guests keep the domain as passed by XAPI (an integer).
+ *)
+let domain_of ~dp ~vm =
+  let vm = Storage_interface.Vm.string_of vm in
+  match vm with
+  | "0" ->
+      (* SM tries to use this in filesystem paths, so cannot have /,
+         and systemd might be a bit unhappy with - *)
+      "u0-" ^ dp |> String.map (function '/' | '-' -> '_' | c -> c)
+  | _ ->
+      vm
+
 (** Functions to wrap calls to the above client modules and convert their
     exceptions and errors into SMAPIv2 errors of type
     [Storage_interface.Exception.exnty]. The above client modules should only
@@ -460,6 +477,8 @@ let fork_exec_rpc :
     )
     >>>= fun input ->
     let input = compat_in input |> Jsonrpc.to_string in
+    debug (fun m -> m "Running %s" @@ Filename.quote_command script_name args)
+    >>= fun () ->
     Process.run ~env ~prog:script_name ~args ~input >>= fun output ->
     let fail_because ~cause description =
       fail
@@ -483,12 +502,13 @@ let fork_exec_rpc :
       with
       | Error _ ->
           error (fun m ->
-              m "%s failed and printed bad error json: %s" script_name
-                output.Process.Output.stdout
+              m "%s[%d] failed and printed bad error json: %s" script_name
+                output.pid output.Process.Output.stdout
           )
           >>= fun () ->
           error (fun m ->
-              m "%s failed, stderr: %s" script_name output.Process.Output.stderr
+              m "%s[%d] failed, stderr: %s" script_name output.pid
+                output.Process.Output.stderr
           )
           >>= fun () ->
           fail_because "non-zero exit and bad json on stdout"
@@ -499,12 +519,12 @@ let fork_exec_rpc :
         with
         | Error _ ->
             error (fun m ->
-                m "%s failed and printed bad error json: %s" script_name
-                  output.Process.Output.stdout
+                m "%s[%d] failed and printed bad error json: %s" script_name
+                  output.pid output.Process.Output.stdout
             )
             >>= fun () ->
             error (fun m ->
-                m "%s failed, stderr: %s" script_name
+                m "%s[%d] failed, stderr: %s" script_name output.pid
                   output.Process.Output.stderr
             )
             >>= fun () ->
@@ -515,7 +535,9 @@ let fork_exec_rpc :
       )
     )
     | Error (Signal signal) ->
-        error (fun m -> m "%s caught a signal and failed" script_name)
+        error (fun m ->
+            m "%s[%d] caught a signal and failed" script_name output.pid
+        )
         >>= fun () -> fail_because "signalled" ~cause:(Signal.to_string signal)
     | Ok () -> (
       (* Parse the json on stdout. We get back a JSON-RPC
@@ -527,8 +549,8 @@ let fork_exec_rpc :
       with
       | Error _ ->
           error (fun m ->
-              m "%s succeeded but printed bad json: %s" script_name
-                output.Process.Output.stdout
+              m "%s[%d] succeeded but printed bad json: %s" script_name
+                output.pid output.Process.Output.stdout
           )
           >>= fun () ->
           fail
@@ -537,7 +559,8 @@ let fork_exec_rpc :
             )
       | Ok response ->
           info (fun m ->
-              m "%s succeeded: %s" script_name output.Process.Output.stdout
+              m "%s[%d] succeeded: %s" script_name output.pid
+                output.Process.Output.stdout
           )
           >>= fun () ->
           let response = compat_out response in
@@ -744,7 +767,7 @@ let vdi_of_volume x =
   ; persistent= true
   }
 
-let choose_datapath ?(persistent = true) domain response =
+let choose_datapath ?(persistent = true) response =
   (* We can only use a URI with a valid scheme, since we use the scheme
      to name the datapath plugin. *)
   let possible =
@@ -789,7 +812,7 @@ let choose_datapath ?(persistent = true) domain response =
   | [] ->
       fail (missing_uri ())
   | (script_dir, scheme, u) :: _us ->
-      return (fork_exec_rpc ~script_dir, scheme, u, domain)
+      return (fork_exec_rpc ~script_dir, scheme, u)
 
 (* Bind the implementations *)
 let bind ~volume_script_dir =
@@ -863,7 +886,7 @@ let bind ~volume_script_dir =
         stat ~dbg ~sr ~vdi:temporary
     )
     >>>= fun response ->
-    choose_datapath domain response >>>= fun (rpc, _datapath, uri, domain) ->
+    choose_datapath response >>>= fun (rpc, _datapath, uri) ->
     return_data_rpc (fun () -> Datapath_client.attach (rpc ~dbg) dbg uri domain)
   in
   let wrap th = Rpc_lwt.T.put th in
@@ -1432,9 +1455,9 @@ let bind ~volume_script_dir =
     |> wrap
   in
   S.VDI.introduce vdi_introduce_impl ;
-  let vdi_attach3_impl dbg _dp sr vdi' vm' _readwrite =
+  let vdi_attach3_impl dbg dp sr vdi' vm _readwrite =
     (let vdi = Storage_interface.Vdi.string_of vdi' in
-     let domain = Storage_interface.Vm.string_of vm' in
+     let domain = domain_of ~dp ~vm in
      vdi_attach_common dbg sr vdi domain >>>= fun response ->
      let convert_implementation = function
        | Xapi_storage.Data.XenDisk {params; extra; backend_type} ->
@@ -1456,9 +1479,9 @@ let bind ~volume_script_dir =
     |> wrap
   in
   S.VDI.attach3 vdi_attach3_impl ;
-  let vdi_activate_common dbg sr vdi' vm' readonly =
+  let vdi_activate_common dbg dp sr vdi' vm readonly =
     (let vdi = Storage_interface.Vdi.string_of vdi' in
-     let domain = Storage_interface.Vm.string_of vm' in
+     let domain = domain_of ~dp ~vm in
      Attached_SRs.find sr >>>= fun sr ->
      (* Discover the URIs using Volume.stat *)
      stat ~dbg ~sr ~vdi >>>= fun response ->
@@ -1472,7 +1495,7 @@ let bind ~volume_script_dir =
          stat ~dbg ~sr ~vdi:temporary
      )
      >>>= fun response ->
-     choose_datapath domain response >>>= fun (rpc, _datapath, uri, domain) ->
+     choose_datapath response >>>= fun (rpc, _datapath, uri) ->
      return_data_rpc (fun () ->
          let rpc = rpc ~dbg in
          if readonly then
@@ -1483,17 +1506,17 @@ let bind ~volume_script_dir =
     )
     |> wrap
   in
-  let vdi_activate3_impl dbg _dp sr vdi' vm' =
-    vdi_activate_common dbg sr vdi' vm' false
+  let vdi_activate3_impl dbg dp sr vdi' vm =
+    vdi_activate_common dbg dp sr vdi' vm false
   in
   S.VDI.activate3 vdi_activate3_impl ;
-  let vdi_activate_readonly_impl dbg _dp sr vdi' vm' =
-    vdi_activate_common dbg sr vdi' vm' true
+  let vdi_activate_readonly_impl dbg dp sr vdi' vm =
+    vdi_activate_common dbg dp sr vdi' vm true
   in
   S.VDI.activate_readonly vdi_activate_readonly_impl ;
-  let vdi_deactivate_impl dbg _dp sr vdi' vm' =
+  let vdi_deactivate_impl dbg dp sr vdi' vm =
     (let vdi = Storage_interface.Vdi.string_of vdi' in
-     let domain = Storage_interface.Vm.string_of vm' in
+     let domain = domain_of ~dp ~vm in
      Attached_SRs.find sr >>>= fun sr ->
      (* Discover the URIs using Volume.stat *)
      stat ~dbg ~sr ~vdi >>>= fun response ->
@@ -1506,17 +1529,17 @@ let bind ~volume_script_dir =
          stat ~dbg ~sr ~vdi:temporary
      )
      >>>= fun response ->
-     choose_datapath domain response >>>= fun (rpc, _datapath, uri, domain) ->
+     choose_datapath response >>>= fun (rpc, _datapath, uri) ->
      return_data_rpc (fun () ->
          Datapath_client.deactivate (rpc ~dbg) dbg uri domain
      )
     )
     |> wrap
   in
   S.VDI.deactivate vdi_deactivate_impl ;
-  let vdi_detach_impl dbg _dp sr vdi' vm' =
+  let vdi_detach_impl dbg dp sr vdi' vm =
     (let vdi = Storage_interface.Vdi.string_of vdi' in
-     let domain = Storage_interface.Vm.string_of vm' in
+     let domain = domain_of ~dp ~vm in
      Attached_SRs.find sr >>>= fun sr ->
      (* Discover the URIs using Volume.stat *)
      stat ~dbg ~sr ~vdi >>>= fun response ->
@@ -1529,7 +1552,7 @@ let bind ~volume_script_dir =
          stat ~dbg ~sr ~vdi:temporary
      )
      >>>= fun response ->
-     choose_datapath domain response >>>= fun (rpc, _datapath, uri, domain) ->
+     choose_datapath response >>>= fun (rpc, _datapath, uri) ->
      return_data_rpc (fun () -> Datapath_client.detach (rpc ~dbg) dbg uri domain)
     )
     |> wrap
@@ -1564,14 +1587,12 @@ let bind ~volume_script_dir =
     |> wrap
   in
   S.SR.stat sr_stat_impl ;
-  let vdi_epoch_begin_impl dbg sr vdi' vm' persistent =
+  let vdi_epoch_begin_impl dbg sr vdi' _vm persistent =
     (let vdi = Storage_interface.Vdi.string_of vdi' in
-     let domain = Storage_interface.Vm.string_of vm' in
      Attached_SRs.find sr >>>= fun sr ->
      (* Discover the URIs using Volume.stat *)
      stat ~dbg ~sr ~vdi >>>= fun response ->
-     choose_datapath ~persistent domain response
-     >>>= fun (rpc, datapath, uri, _domain) ->
+     choose_datapath ~persistent response >>>= fun (rpc, datapath, uri) ->
      (* If non-persistent and the datapath plugin supports NONPERSISTENT
         then we delegate this to the datapath plugin. Otherwise we will
         make a temporary clone now and attach/detach etc this file. *)
@@ -1602,13 +1623,12 @@ let bind ~volume_script_dir =
     |> wrap
   in
   S.VDI.epoch_begin vdi_epoch_begin_impl ;
-  let vdi_epoch_end_impl dbg sr vdi' vm' =
+  let vdi_epoch_end_impl dbg sr vdi' _vm =
     (let vdi = Storage_interface.Vdi.string_of vdi' in
-     let domain = Storage_interface.Vm.string_of vm' in
      Attached_SRs.find sr >>>= fun sr ->
      (* Discover the URIs using Volume.stat *)
      stat ~dbg ~sr ~vdi >>>= fun response ->
-     choose_datapath domain response >>>= fun (rpc, datapath, uri, _domain) ->
+     choose_datapath response >>>= fun (rpc, datapath, uri) ->
      if Datapath_plugins.supports_feature datapath _nonpersistent then
        return_data_rpc (fun () -> Datapath_client.close (rpc ~dbg) dbg uri)
      else
@@ -1627,9 +1647,9 @@ let bind ~volume_script_dir =
   S.VDI.epoch_end vdi_epoch_end_impl ;
   let vdi_set_persistent_impl _dbg _sr _vdi _persistent = return () |> wrap in
   S.VDI.set_persistent vdi_set_persistent_impl ;
-  let dp_destroy2 dbg _dp sr vdi' vm' _allow_leak =
+  let dp_destroy2 dbg dp sr vdi' vm _allow_leak =
     (let vdi = Storage_interface.Vdi.string_of vdi' in
-     let domain = Storage_interface.Vm.string_of vm' in
+     let domain = domain_of ~dp ~vm in
      Attached_SRs.find sr >>>= fun sr ->
      (* Discover the URIs using Volume.stat *)
      stat ~dbg ~sr ~vdi >>>= fun response ->
@@ -1642,7 +1662,7 @@ let bind ~volume_script_dir =
          stat ~dbg ~sr ~vdi:temporary
      )
      >>>= fun response ->
-     choose_datapath domain response >>>= fun (rpc, _datapath, uri, domain) ->
+     choose_datapath response >>>= fun (rpc, _datapath, uri) ->
      return_data_rpc (fun () ->
          Datapath_client.deactivate (rpc ~dbg) dbg uri domain
      )

ocaml/xapi-storage-script/test_lib.ml

@@ -103,12 +103,20 @@ let test_run_status =
   let module P = Process in
   let test () =
     let* output = P.run ~prog:"true" ~args:[] ~input:"" ~env:[] in
-    let expected = P.Output.{exit_status= Ok (); stdout= ""; stderr= ""} in
+    let expected =
+      P.Output.{exit_status= Ok (); pid= output.pid; stdout= ""; stderr= ""}
+    in
     Alcotest.(check output_c) "Exit status is correct" expected output ;
 
     let* output = P.run ~prog:"false" ~args:[] ~input:"" ~env:[] in
     let expected =
-      P.Output.{exit_status= Error (Exit_non_zero 1); stdout= ""; stderr= ""}
+      P.Output.
+        {
+          exit_status= Error (Exit_non_zero 1)
+        ; pid= output.pid
+        ; stdout= ""
+        ; stderr= ""
+        }
     in
     Alcotest.(check output_c) "Exit status is correct" expected output ;
 
@@ -121,15 +129,24 @@ let test_run_output =
   let test () =
     let content = "@@@@@@" in
     let* output = P.run ~prog:"cat" ~args:["-"] ~input:content ~env:[] in
-    let expected = P.Output.{exit_status= Ok (); stdout= content; stderr= ""} in
+    let expected =
+      P.Output.
+        {exit_status= Ok (); pid= output.pid; stdout= content; stderr= ""}
+    in
     Alcotest.(check output_c) "Stdout is correct" expected output ;
 
     let* output = P.run ~prog:"cat" ~args:[content] ~input:content ~env:[] in
     let stderr =
       Printf.sprintf "cat: %s: No such file or directory\n" content
     in
     let expected =
-      P.Output.{exit_status= Error (Exit_non_zero 1); stdout= ""; stderr}
+      P.Output.
+        {
+          exit_status= Error (Exit_non_zero 1)
+        ; pid= output.pid
+        ; stdout= ""
+        ; stderr
+        }
     in
     Alcotest.(check output_c) "Stderr is correct" expected output ;
     Lwt.return ()