stream_vdi: Only process allocated clusters for VHD and QCOW on XVA export

dune-project

@@ -536,7 +536,8 @@
   (xen-api-client-lwt
    (= :version))
   xenstore
-  xenstore_transport))
+  xenstore_transport
+  yojson))
 
 (package
  (name vhd-format))

ocaml/libs/vhd/vhd_format/dune

@@ -2,5 +2,6 @@
  (name vhd_format)
  (public_name vhd-format)
  (flags :standard -w -32-34-37)
- (libraries stdlib-shims (re_export bigarray-compat) cstruct io-page rresult unix uuidm)
+ (libraries stdlib-shims (re_export bigarray-compat) cstruct io-page rresult
+            unix uuidm yojson)
  (preprocess (pps ppx_cstruct)))

ocaml/libs/vhd/vhd_format/f.ml

@@ -2883,6 +2883,37 @@ functor
 
       let raw ?from (vhd : fd Vhd.t) = raw_common ?from vhd
 
+      let vhd_blocks_to_json (t : fd Vhd.t) =
+        let block_size_sectors_shift =
+          t.Vhd.header.Header.block_size_sectors_shift
+        in
+        let max_table_entries = Vhd.used_max_table_entries t in
+
+        let include_block = include_block None t in
+
+        let blocks =
+          Seq.init max_table_entries Fun.id
+          |> Seq.filter_map (fun i ->
+                 if include_block i then
+                   Some (`Int i)
+                 else
+                   None
+             )
+          |> List.of_seq
+        in
+        let json =
+          `Assoc
+            [
+              ( "virtual_size"
+              , `Int (Int64.to_int t.Vhd.footer.Footer.current_size)
+              )
+            ; ("cluster_bits", `Int (block_size_sectors_shift + sector_shift))
+            ; ("data_clusters", `List blocks)
+            ]
+        in
+        let json_string = Yojson.to_string json in
+        print_string json_string ; return ()
+
       let vhd_common ?from ?raw ?(emit_batmap = false) (t : fd Vhd.t) =
         let block_size_sectors_shift =
           t.Vhd.header.Header.block_size_sectors_shift
@@ -3119,6 +3150,8 @@ functor
 
       let vhd ?from (raw : 'a) (vhd : fd Vhd.t) =
         Vhd_input.vhd_common ?from ~raw vhd
+
+      let blocks_json = Vhd_input.vhd_blocks_to_json
     end
 
     (* Create a VHD stream from data on t, using `include_block` guide us which blocks have data *)

ocaml/libs/vhd/vhd_format/f.mli

@@ -474,6 +474,8 @@ module From_file : functor (F : S.FILE) -> sig
         copies from the virtual disk [raw]. If [from] is provided then the
         stream will contain only the virtual updates required to transform
         [from] into [t] *)
+
+    val blocks_json : fd Vhd.t -> unit t
   end
 
   module Raw_input : sig

ocaml/tapctl/tapctl.ml

@@ -535,9 +535,9 @@ let of_device ctx path =
   if driver_of_major major <> "tapdev" then raise Not_blktap ;
   match List.filter (fun (tapdev, _, _) -> tapdev.minor = minor) (list ctx) with
   | [t] ->
-      t
+      Some t
   | _ ->
-      raise Not_found
+      None
 
 let find ctx ~pid ~minor =
   match list ~t:{minor; tapdisk_pid= pid} ctx with

ocaml/tapctl/tapctl.mli

@@ -91,7 +91,7 @@ exception Not_blktap
 (** Thrown by [of_device x] when [x] is not a device *)
 exception Not_a_device
 
-val of_device : context -> string -> t
+val of_device : context -> string -> t option
 (** Given a path to a device, return the corresponding tap information *)
 
 val find : context -> pid:int -> minor:int -> t

ocaml/vhd-tool/cli/main.ml

@@ -385,9 +385,29 @@ let stream_cmd =
   , Cmd.info "stream" ~sdocs:_common_options ~doc ~man
   )
 
+let read_headers_cmd =
+  let doc =
+    "Parse VHD headers and output allocated blocks information in JSON format \
+     like: {'virtual_size': X, 'cluster_bits: X, 'data_clusters: [1,2,3]}"
+  in
+  let source =
+    let doc = Printf.sprintf "Path to the VHD file" in
+    Arg.(required & pos 0 (some file) None & info [] ~doc)
+  in
+  ( Term.(ret (const Impl.read_headers $ common_options_t $ source))
+  , Cmd.info "read_headers" ~sdocs:_common_options ~doc
+  )
+
 let cmds =
   [
-    info_cmd; contents_cmd; get_cmd; create_cmd; check_cmd; serve_cmd; stream_cmd
+    info_cmd
+  ; contents_cmd
+  ; get_cmd
+  ; create_cmd
+  ; check_cmd
+  ; serve_cmd
+  ; stream_cmd
+  ; read_headers_cmd
   ]
   |> List.map (fun (t, i) -> Cmd.v i t)
 

ocaml/vhd-tool/cli/sparse_dd.ml

@@ -244,14 +244,14 @@ let with_paused_tapdisk path f =
   let path = find_backend_device path |> Opt.default path in
   let context = Tapctl.create () in
   match Tapctl.of_device context path with
-  | tapdev, _, Some (_driver, path) ->
+  | Some (tapdev, _, Some (_driver, path)) ->
       debug "pausing tapdisk for %s" path ;
       Tapctl.pause context tapdev ;
       after f (fun () ->
           debug "unpausing tapdisk for %s" path ;
           Tapctl.unpause context tapdev path Tapctl.Vhd
       )
-  | _, _, _ ->
+  | _ ->
       failwith (Printf.sprintf "Failed to pause tapdisk for %s" path)
 
 (* Record when the binary started for performance measuring *)

ocaml/vhd-tool/src/image.ml

@@ -66,11 +66,11 @@ let image_behind_nbd_device image =
 
 let of_device path =
   match Tapctl.of_device (Tapctl.create ()) path with
-  | _, _, Some ("vhd", vhd) ->
+  | Some (_, _, Some ("vhd", vhd)) ->
       Some (`Vhd vhd)
-  | _, _, Some ("aio", vhd) ->
+  | Some (_, _, Some ("aio", vhd)) ->
       Some (`Raw vhd)
-  | _, _, _ ->
+  | _ ->
       None
   | exception Tapctl.Not_blktap ->
       get_nbd_device path |> image_behind_nbd_device

ocaml/vhd-tool/src/impl.ml

@@ -1164,6 +1164,14 @@ let stream_t common args ?(progress = no_progress_bar) () =
     args.StreamCommon.tar_filename_prefix args.StreamCommon.good_ciphersuites
     args.StreamCommon.verify_cert
 
+let read_headers common source =
+  let path = [Filename.dirname source] in
+  let thread =
+    retry common 3 (fun () -> Vhd_IO.openchain ~path source false) >>= fun t ->
+    Vhd_IO.close t >>= fun () -> Hybrid_input.blocks_json t
+  in
+  Lwt_main.run thread ; `Ok ()
+
 let stream common args =
   try
     Vhd_format_lwt.File.use_unbuffered := common.Common.unbuffered ;

ocaml/vhd-tool/src/impl.mli

@@ -35,6 +35,9 @@ val check :
 val stream :
   Common.t -> StreamCommon.t -> [> `Error of bool * string | `Ok of unit]
 
+val read_headers :
+  Common.t -> string -> [> `Error of bool * string | `Ok of unit]
+
 val serve :
      Common.t
   -> string

ocaml/xapi/dune

@@ -259,6 +259,7 @@
     Storage_mux
     Storage_smapiv1_wrapper
     Stream_vdi
+    Xapi_vdi_helpers
     System_domains
     Xapi_psr
     Xapi_services

ocaml/xapi/qcow_tool_wrapper.ml

@@ -12,44 +12,14 @@
  * GNU Lesser General Public License for more details.
  *)
 
-module D = Debug.Make (struct let name = __MODULE__ end)
-
-open D
-
-let run_qcow_tool qcow_tool ?(replace_fds = []) ?input_fd ?output_fd
-    (_progress_cb : int -> unit) (args : string list) =
-  info "Executing %s %s" qcow_tool (String.concat " " args) ;
-  let open Forkhelpers in
-  match
-    with_logfile_fd "qcow-tool" (fun log_fd ->
-        let pid =
-          safe_close_and_exec input_fd output_fd (Some log_fd) replace_fds
-            qcow_tool args
-        in
-        let _, status = waitpid pid in
-        if status <> Unix.WEXITED 0 then (
-          error "qcow-tool failed, returning VDI_IO_ERROR" ;
-          raise
-            (Api_errors.Server_error
-               (Api_errors.vdi_io_error, ["Device I/O errors"])
-            )
-        )
-    )
-  with
-  | Success (out, _) ->
-      debug "qcow-tool successful export (%s)" out
-  | Failure (out, _e) ->
-      error "qcow-tool output: %s" out ;
-      raise (Api_errors.Server_error (Api_errors.vdi_io_error, [out]))
-
 let update_task_progress (__context : Context.t) (x : int) =
   TaskHelper.set_progress ~__context (float_of_int x /. 100.)
 
 let receive (progress_cb : int -> unit) (unix_fd : Unix.file_descr)
     (path : string) =
   let args = ["stream_decode"; path] in
   let qcow_tool = !Xapi_globs.qcow_stream_tool in
-  run_qcow_tool qcow_tool progress_cb args ~input_fd:unix_fd
+  Vhd_qcow_parsing.run_tool qcow_tool progress_cb args ~input_fd:unix_fd
 
 let read_header qcow_path =
   let args = ["read_headers"; qcow_path] in
@@ -58,28 +28,21 @@ let read_header qcow_path =
 
   let progress_cb _ = () in
   Xapi_stdext_pervasives.Pervasiveext.finally
-    (fun () -> run_qcow_tool qcow_tool progress_cb args ~output_fd:pipe_writer)
+    (fun () ->
+      Vhd_qcow_parsing.run_tool qcow_tool progress_cb args
+        ~output_fd:pipe_writer
+    )
     (fun () -> Unix.close pipe_writer) ;
   pipe_reader
 
 let parse_header qcow_path =
   let pipe_reader = read_header qcow_path in
-  let ic = Unix.in_channel_of_descr pipe_reader in
-  let buf = Buffer.create 4096 in
-  let json = Yojson.Basic.from_channel ~buf ~fname:"qcow_header.json" ic in
-  In_channel.close ic ;
-  let cluster_size =
-    1 lsl Yojson.Basic.Util.(member "cluster_bits" json |> to_int)
-  in
-  let cluster_list =
-    Yojson.Basic.Util.(member "data_clusters" json |> to_list |> List.map to_int)
-  in
-  (cluster_size, cluster_list)
+  Vhd_qcow_parsing.parse_header pipe_reader
 
 let send ?relative_to (progress_cb : int -> unit) (unix_fd : Unix.file_descr)
     (path : string) (_size : Int64.t) =
   let qcow_of_device =
-    Vhd_tool_wrapper.backing_file_of_device ~driver:"qcow2"
+    Xapi_vdi_helpers.backing_file_of_device_with_driver ~driver:"qcow2"
   in
   let qcow_path = qcow_of_device path in
 
@@ -107,8 +70,8 @@ let send ?relative_to (progress_cb : int -> unit) (unix_fd : Unix.file_descr)
   let replace_fds = Option.map (fun fd -> [(unique_string, fd)]) diff_fd in
   Xapi_stdext_pervasives.Pervasiveext.finally
     (fun () ->
-      run_qcow_tool qcow_tool progress_cb args ?input_fd ~output_fd:unix_fd
-        ?replace_fds
+      Vhd_qcow_parsing.run_tool qcow_tool progress_cb args ?input_fd
+        ~output_fd:unix_fd ?replace_fds
     )
     (fun () ->
       Option.iter Unix.close input_fd ;

ocaml/xapi/storage_smapiv1_migrate.ml

@@ -106,18 +106,16 @@ let tapdisk_of_attach_info (backend : Storage_interface.backend) =
   match (blockdevices, nbds) with
   | blockdevice :: _, _ -> (
       let path = blockdevice.Storage_interface.path in
-      try
-        match Tapctl.of_device (Tapctl.create ()) path with
-        | tapdev, _, _ ->
-            Some tapdev
-      with
-      | Tapctl.Not_blktap ->
+      match Tapctl.of_device (Tapctl.create ()) path with
+      | Some (tapdev, _, _) ->
+          Some tapdev
+      | exception Tapctl.Not_blktap ->
           D.debug "Device %s is not controlled by blktap" path ;
           None
-      | Tapctl.Not_a_device ->
+      | exception Tapctl.Not_a_device ->
           D.debug "%s is not a device" path ;
           None
-      | _ ->
+      | (exception _) | None ->
           D.debug "Device %s has an unknown driver" path ;
           None
     )
@@ -295,8 +293,8 @@ module Copy = struct
       perform_cleanup_actions !on_fail ;
       raise e
 
-  (** [copy_into_sr] does not requires a dest vdi to be provided, instead, it will 
-  find the nearest vdi on the [dest] sr, and if there is no such vdi, it will 
+  (** [copy_into_sr] does not requires a dest vdi to be provided, instead, it will
+  find the nearest vdi on the [dest] sr, and if there is no such vdi, it will
   create one. *)
   let copy_into_sr ~task ~dbg ~sr ~vdi ~vm ~url ~dest ~verify_dest =
     D.debug "copy sr:%s vdi:%s url:%s dest:%s verify_dest:%B"