Add optional compression to VM exports. Both compressed and uncompressed formats...

Design notes on the wiki: http://wiki.xensource.com/xenwiki/Compressing_VM_exports

Signed-off-by: David Scott <dave.scott@eu.citrix.com>

diff --git a/ocaml/idl/constants.ml b/ocaml/idl/constants.ml
index 86dc782225f2ec1cb5f651ff1ca0c6a816858b4c..19ce572c1efd733321f86c2f37bf2787eda23928 100644 (file)
--- a/ocaml/idl/constants.ml
+++ b/ocaml/idl/constants.ml
@@ -47,6 +47,8 @@ let wlb_report_uri = "/wlb_report"                    (* ocaml/xapi/wlb_reports.
 let wlb_diagnostics_uri = "/wlb_diagnostics"          (* ocaml/xapi/wlb_reports.ml *)
 let audit_log_uri = "/audit_log"                      (* ocaml/xapi/audit.ml *)
 
+let use_compression = "use_compression"
+
 (* If VM.HVM_boot_policy is set to this then we boot using qemu-dm *)
 let hvm_boot_policy_bios_order = "BIOS order"
 (* Key we expect to find in VM.HVM_boot_params if VM.HVM_boot_policy = BIOS_order.

diff --git a/ocaml/xapi/cli_frontend.ml b/ocaml/xapi/cli_frontend.ml
index 3f8d8d5a2416bce3e3185325123b488ef17eade6..4780599a5a6c7cfdc29f69e7d08db8e2255ed84d 100644 (file)
--- a/ocaml/xapi/cli_frontend.ml
+++ b/ocaml/xapi/cli_frontend.ml
@@ -1231,7 +1231,7 @@ there are two or more empty CD devices, please use the command 'vbd-insert' and
    "vm-export",
     {
       reqd=["filename"];
-      optn=["preserve-power-state"];
+      optn=["preserve-power-state"; "compress"];
       help="Export a VM to <filename>.";
       implementation= With_fd Cli_operations.vm_export;
       flags=[Standard; Vm_selectors];

diff --git a/ocaml/xapi/cli_operations.ml b/ocaml/xapi/cli_operations.ml
index d1cc85def7d14635beea39a37a8259dd86a09711..381f43133ae647170ae25d334ca2920a010009eb 100644 (file)
--- a/ocaml/xapi/cli_operations.ml
+++ b/ocaml/xapi/cli_operations.ml
@@ -3067,7 +3067,7 @@ let blob_create printer rpc session_id params =
     raise (Cli_util.Cli_failure "Need one of: vm-uuid, host-uuid, network-uuid, sr-uuid or pool-uuid")
 
     
-let export_common fd printer rpc session_id params filename num preserve_power_state vm =
+let export_common fd printer rpc session_id params filename num use_compression preserve_power_state vm =
   let vm_record = vm.record () in
   let exporttask = Client.Task.create rpc session_id (Printf.sprintf "Export of VM: %s" (vm_record.API.vM_uuid)) "" in
   
@@ -3082,13 +3082,15 @@ let export_common fd printer rpc session_id params filename num preserve_power_s
       let f = if !num > 1 then filename ^ (string_of_int !num) else filename in
       download_file ~__context rpc session_id exporttask fd f
         (Printf.sprintf
-           "%s?session_id=%s&task_id=%s&ref=%s&preserve_power_state=%b" 
+           "%s?session_id=%s&task_id=%s&ref=%s&%s=%s&preserve_power_state=%b" 
           (if List.mem_assoc "metadata" params
            then Constants.export_metadata_uri
            else Constants.export_uri)
           (Ref.string_of session_id)
            (Ref.string_of exporttask) 
           (Ref.string_of (vm.getref ()))
+          Constants.use_compression
+          (if use_compression then "true" else "false")
                preserve_power_state)
         "Export";
         num := !num + 1)
@@ -3096,20 +3098,22 @@ let export_common fd printer rpc session_id params filename num preserve_power_s
     
 let vm_export fd printer rpc session_id params =
   let filename = List.assoc "filename" params in
+  let use_compression = List.mem_assoc "compress" params && (List.assoc "compress" params = "true") in
   let preserve_power_state = List.mem_assoc "preserve-power-state" params && bool_of_string "preserve-power-state" (List.assoc "preserve-power-state" params) in
   let num = ref 1 in
   let op vm = 
-    export_common fd printer rpc session_id params filename num preserve_power_state vm
+    export_common fd printer rpc session_id params filename num use_compression preserve_power_state vm
   in
-  ignore(do_vm_op printer rpc session_id op params ["filename"; "metadata"; "preserve-power-state"])
+  ignore(do_vm_op printer rpc session_id op params ["filename"; "metadata"; "compress"; "preserve-power-state"])
 
 let vm_export_aux obj_type fd printer rpc session_id params =
   let filename = List.assoc "filename" params in
+  let use_compression = List.mem_assoc "compress" params && (List.assoc "compress" params = "true") in
   let preserve_power_state = List.mem_assoc "preserve-power-state" params && bool_of_string "preserve-power-state" (List.assoc "preserve-power-state" params) in
   let num = ref 1 in
   let uuid = List.assoc (obj_type ^ "-uuid") params in
   let ref = Client.VM.get_by_uuid rpc session_id uuid in
-  export_common fd printer rpc session_id params filename num preserve_power_state (vm_record rpc session_id ref)
+  export_common fd printer rpc session_id params filename num use_compression preserve_power_state (vm_record rpc session_id ref)
 
 let vm_copy_bios_strings printer rpc session_id params =
   let host = Client.Host.get_by_uuid rpc session_id (List.assoc "host-uuid" params) in

diff --git a/ocaml/xapi/export.ml b/ocaml/xapi/export.ml
index aed8a78552116955fc804ee725bf6e8354e3e26a..259056d26f19f8d01241d2e821e6dd1d6dfd5031 100644 (file)
--- a/ocaml/xapi/export.ml
+++ b/ocaml/xapi/export.ml
@@ -419,6 +419,8 @@ let handler (req: request) s =
   
   Xapi_http.assert_credentials_ok "VM.export" ~http_action:"get_export" req;
     
+  let use_compression = List.mem_assoc Constants.use_compression req.query && List.assoc Constants.use_compression req.query = "true" in
+  debug "Using compression: %b" use_compression;
   (* Perform the SR reachability check using a fresh context/task because
      we don't want to complete the task in the forwarding case *)
   
@@ -485,7 +487,11 @@ let handler (req: request) s =
               with_vm_locked ~__context ~vm:vm_ref ~task_id `export
                 (fun () -> 
                    Http_svr.headers s headers;
-                   export refresh_session __context rpc session_id s vm_ref preserve_power_state)
+                   let go fd = export refresh_session __context rpc session_id fd vm_ref preserve_power_state in
+                   if use_compression
+                   then Gzip.compress s go
+                   else go s
+                )
                 
                     (* Exceptions are handled by Server_helpers.with_context *)
             ))

diff --git a/ocaml/xapi/import.ml b/ocaml/xapi/import.ml
index 961abf2063cd1baf2e173dc917319382afc201e0..bc47e3257fb356cb545763fd8e26a47d8b86da9d 100644 (file)
--- a/ocaml/xapi/import.ml
+++ b/ocaml/xapi/import.ml
@@ -576,17 +576,66 @@ let handle_all __context config rpc session_id (xs: obj list) =
     end;
     raise e
 
-(** Read the next file from the tar stream as XML metadata *)
-let get_xml fd filename =
-  (* Read the xml header *)
-  let xml = Tar.Archive.with_next_file fd
-    (fun s hdr ->
-       if hdr.Tar.Header.file_name <> filename then raise (IFailure (Unexpected_file (filename, hdr.Tar.Header.file_name)));
-       let file_size = hdr.Tar.Header.file_size in
-       let xml_string = Bigbuffer.make () in
-       really_read_bigbuffer s xml_string file_size;
-       xml_string) in
-  Xml.parse_bigbuffer xml
+(** Read the next file in the archive as xml *)
+let read_xml hdr fd = 
+  let xml_string = Bigbuffer.make () in
+  really_read_bigbuffer fd xml_string hdr.Tar.Header.file_size;
+  Xml.parse_bigbuffer xml_string
+
+let assert_filename_is hdr filename = 
+  if hdr.Tar.Header.file_name <> filename then begin
+    let hex = Tar.Header.to_hex in
+    error "import expects the next file in the stream to be [%s]; got [%s]"
+      (hex  hdr.Tar.Header.file_name) (hex Xva.xml_filename);
+    raise (IFailure (Unexpected_file(hdr.Tar.Header.file_name, Xva.xml_filename)))
+  end
+
+(** Takes an fd and a function, tries first to read the first tar block
+    and checks for the existence of 'ova.xml'. If that fails then pipe
+    the lot through gzip and try again *)
+let with_open_archive fd f = 
+  (* Read the first header's worth into a buffer *)
+  let buffer = String.make Tar.Header.length ' ' in
+  let retry_with_gzip = ref true in
+  try
+    really_read fd buffer 0 Tar.Header.length;
+
+    (* we assume the first block is not all zeroes *)
+    let Some hdr = Tar.Header.unmarshal buffer in
+    assert_filename_is hdr Xva.xml_filename;
+
+    (* successfully opened uncompressed stream *)
+    retry_with_gzip := false;
+    let xml = read_xml hdr fd in
+    Tar.Archive.skip fd (Tar.Header.compute_zero_padding_length hdr);
+    f xml fd
+  with e ->
+    if not(!retry_with_gzip) then raise e;
+    debug "Failed to directly open the archive; trying gzip";
+    let pipe_out, pipe_in = Unix.pipe () in
+    let t = Thread.create
+      (Gzip.decompress pipe_in)
+      (fun compressed_in ->
+        (* Write the initial buffer *)
+        Unix.set_close_on_exec compressed_in;
+        debug "Writing initial buffer";
+        Unix.write compressed_in buffer 0 Tar.Header.length;
+        let n = Unixext.copy_file fd compressed_in in
+        debug "Written a total of %d + %Ld bytes" Tar.Header.length n;
+      ) in
+    finally
+      (fun () ->
+        let hdr = Tar.Header.get_next_header pipe_out in
+        assert_filename_is hdr Xva.xml_filename;
+
+        let xml = read_xml hdr pipe_out in
+        Tar.Archive.skip pipe_out (Tar.Header.compute_zero_padding_length hdr);
+        f xml pipe_out)
+      (fun () ->
+        debug "Closing pipes";
+        Unix.close pipe_in;
+        Unix.close pipe_out;
+        Thread.join t)
 
 (** Remove "import" from the current operations of all created VMs, complete the
     task including the VM references *)
@@ -622,7 +671,9 @@ let metadata_handler (req: request) s =
         [ Http.task_id_hdr ^ ":" ^ (Ref.string_of (Context.get_task_id __context));
           content_type ] in
        Http_svr.headers s headers;
-       let metadata = get_xml s Xva.xml_filename in
+       with_open_archive s
+       (fun metadata s ->
+           debug "Got XML";
        (* Skip trailing two zero blocks *)
        Tar.Archive.skip s (Tar.Header.length * 2);
 
@@ -651,7 +702,7 @@ let metadata_handler (req: request) s =
         cleanup on_cleanup_stack;
         end;
         raise e
-    ))
+    )))
 
 let handler (req: request) s = 
   req.close <- true;
@@ -730,8 +781,9 @@ let handler (req: request) s =
                        content_type ] in
                    Http_svr.headers s headers;
                    debug "Reading XML";
-                   let metadata = get_xml s Xva.xml_filename in
-                   debug "Got XML";
+                   with_open_archive s
+                     (fun metadata s ->
+                        debug "Got XML";
                    let old_zurich_or_geneva = try ignore(Xva.of_xml metadata); true with _ -> false in
                    let vmrefs = 
                      if old_zurich_or_geneva 
@@ -769,7 +821,8 @@ let handler (req: request) s =
                          (* against the table here. Nb. Rio GA-Miami B2 exports get their checksums checked twice! *)
                          if header.version.export_vsn < 2 then 
                            begin
-                             let expected_checksums = checksum_table_of_xmlrpc (get_xml s Xva.checksum_filename) in
+                             let xml = Tar.Archive.with_next_file s (fun s hdr -> read_xml hdr s) in
+                             let expected_checksums = checksum_table_of_xmlrpc xml in
                              if not(compare_checksums checksum_table expected_checksums) then begin
                                error "Some data checksums were incorrect: VM may be corrupt";
                                if not(force)
@@ -793,6 +846,7 @@ let handler (req: request) s =
                    in
                    complete_import ~__context vmrefs;
                    debug "import successful"
+                   )
              with 
                | IFailure failure ->
                    begin