Split Storage_migrate.start

The original Storage_migrate.start function is quite fat and does lots
of things. For better readability and maintainability, split it into
multiple functions that represent different stages of the SXM:
- prepare: makes a remote call to the dest host and asks it to prepare
  for the VDIs for receiving data
- mirror_pass_fds: passes the fd to tapdisk to establish connection
  between two tapdisk processes (sending & receiving)
- mirror_snapshot: takes a snapshot of the VDI to be mirrored, because
  the tapdisk mirroring only mirrors data written after the mirror is
  initiated
- mirror_copy: copy the snapshot from the source to destination

As these operations are all specific to SMAPIv1, move them to
`storage_smapiv1_migrate` as well.

Also restructure the way clean up is done. As there are multiple points
at which the migration might fail, previously a `on_fail` list ref is
constructed and populated as we go. Now that we have explicitly defined
the different errors during the migration, we could handle these errors
accordingly.

The way to reason about cleanups is: we have three possible clean up
actions:

- receive_cancel: which will undo the preparation work done on the dest
  host. This is added once preparation is done;
- destroy snapshot: which will destroy the snapshot. This is called
  after the snapshot is created;
- stop: which will disable mirroring, delete snapshot and also do what
  receive_cancel does, which is basically clean up everything. And this
  will be called after the mirror has been started.

Note many of these functions are best-effort, for example, the stop
function won't delete a snapshot if it does not exist, so it's generally
safe to call them.

We put these clean up functions into the appropriate places, see the
actual arrangement of these functions in the code, they should be
self-explanatory.

Signed-off-by: Vincent Liu <shuntian.liu2@cloud.com>

Author: incipit0

ocaml/xapi/storage_migrate.ml

@@ -247,6 +247,20 @@ module MigrateLocal = struct
     | e ->
         raise e
 
+  let prepare ~dbg ~sr ~vdi ~dest ~local_vdi ~mirror_id ~mirror_vm ~url
+      ~verify_dest =
+    try
+      let (module Remote) = get_remote_backend url verify_dest in
+      let similars = similar_vdis ~dbg ~sr ~vdi in
+
+      Remote.DATA.MIRROR.receive_start2 dbg dest local_vdi mirror_id similars
+        mirror_vm
+    with e ->
+      error "%s Caught error %s while preparing for SXM" __FUNCTION__
+        (Printexc.to_string e) ;
+      raise
+        (Storage_error (Migration_preparation_failure (Printexc.to_string e)))
+
   let start ~task ~dbg ~sr ~vdi ~dp ~mirror_vm ~copy_vm ~url ~dest ~verify_dest
       =
     SXM.info
@@ -262,7 +276,6 @@ module MigrateLocal = struct
       (Storage_interface.Sr.string_of dest)
       verify_dest ;
 
-    let remote_url = Http.Url.of_string url in
     let (module Remote) = get_remote_backend url verify_dest in
     (* Find the local VDI *)
     let local_vdi = find_local_vdi ~dbg ~sr ~vdi in
@@ -285,176 +298,51 @@ module MigrateLocal = struct
     State.add mirror_id (State.Send_op alm) ;
     debug "%s Added mirror %s to active local mirrors" __FUNCTION__ mirror_id ;
     (* A list of cleanup actions to perform if the operation should fail. *)
-    let on_fail : (unit -> unit) list ref = ref [] in
     try
-      let similar_vdis = Local.VDI.similar_content dbg sr vdi in
-      let similars =
-        List.filter
-          (fun x -> x <> "")
-          (List.map (fun vdi -> vdi.content_id) similar_vdis)
+      let (Vhd_mirror remote_mirror) =
+        prepare ~dbg ~sr ~vdi ~dest ~local_vdi ~mirror_id ~mirror_vm ~url
+          ~verify_dest
       in
-      debug "Similar VDIs to = [ %s ]"
-        (String.concat "; "
-           (List.map
-              (fun x ->
-                Printf.sprintf "(vdi=%s,content_id=%s)"
-                  (Storage_interface.Vdi.string_of x.vdi)
-                  x.content_id
-              )
-              similar_vdis
-           )
-        ) ;
-      let (Mirror.Vhd_mirror result) =
-        Remote.DATA.MIRROR.receive_start2 dbg dest local_vdi mirror_id similars
-          mirror_vm
-      in
-      (* Enable mirroring on the local machine *)
-      let mirror_dp = result.Mirror.mirror_datapath in
-      let uri =
-        Printf.sprintf "/services/SM/nbd/%s/%s/%s/%s"
-          (Storage_interface.Vm.string_of mirror_vm)
-          (Storage_interface.Sr.string_of dest)
-          (Storage_interface.Vdi.string_of result.Mirror.mirror_vdi.vdi)
-          mirror_dp
-      in
-      debug "%s: uri of http request for mirroring is %s" __FUNCTION__ uri ;
-      let dest_url = Http.Url.set_uri remote_url uri in
-      let request =
-        Http.Request.make
-          ~query:(Http.Url.get_query_params dest_url)
-          ~version:"1.0" ~user_agent:"smapiv2" Http.Put uri
-      in
-      let verify_cert = if verify_dest then Stunnel_client.pool () else None in
-      let transport = Xmlrpc_client.transport_of_url ~verify_cert dest_url in
-      debug "Searching for data path: %s" dp ;
-      let attach_info = Local.DP.attach_info dbg sr vdi dp mirror_vm in
-      on_fail :=
-        (fun () -> Remote.DATA.MIRROR.receive_cancel dbg mirror_id) :: !on_fail ;
       let tapdev =
-        match tapdisk_of_attach_info attach_info with
-        | Some tapdev ->
-            let pid = Tapctl.get_tapdisk_pid tapdev in
-            let path =
-              Printf.sprintf "/var/run/blktap-control/nbdclient%d" pid
-            in
-            with_transport ~stunnel_wait_disconnect:false transport
-              (with_http request (fun (_response, s) ->
-                   let control_fd =
-                     Unix.socket Unix.PF_UNIX Unix.SOCK_STREAM 0
-                   in
-                   finally
-                     (fun () ->
-                       Unix.connect control_fd (Unix.ADDR_UNIX path) ;
-                       let msg = dp in
-                       let len = String.length msg in
-                       let written =
-                         Unixext.send_fd_substring control_fd msg 0 len [] s
-                       in
-                       if written <> len then (
-                         error "Failed to transfer fd to %s" path ;
-                         failwith "Internal error transferring fd to tapdisk"
-                       )
-                     )
-                     (fun () -> Unix.close control_fd)
-               )
-              ) ;
-            tapdev
-        | None ->
-            failwith "Not attached"
-      in
-      debug "%s Updating active local mirrors: id=%s" __FUNCTION__ mirror_id ;
-      let alm =
-        State.Send_state.
-          {
-            url
-          ; dest_sr= dest
-          ; remote_info=
-              Some
-                {
-                  dp= mirror_dp
-                ; vdi= result.Mirror.mirror_vdi.vdi
-                ; url
-                ; verify_dest
-                }
-          ; local_dp= dp
-          ; tapdev= Some tapdev
-          ; failed= false
-          ; watchdog= None
-          }
+        Storage_smapiv1_migrate.mirror_pass_fds ~dbg ~dp ~sr ~vdi ~mirror_vm
+          ~mirror_id ~url ~dest_sr:dest ~verify_dest ~remote_mirror
       in
-
-      State.add mirror_id (State.Send_op alm) ;
-      debug "%s Updated mirror_id %s in the active local mirror" __FUNCTION__
-        mirror_id ;
-
-      SXM.info "%s About to snapshot VDI = %s" __FUNCTION__
-        (string_of_vdi_info local_vdi) ;
-      let local_vdi = add_to_sm_config local_vdi "mirror" ("nbd:" ^ dp) in
-      let local_vdi = add_to_sm_config local_vdi "base_mirror" mirror_id in
       let snapshot =
-        try Local.VDI.snapshot dbg sr local_vdi with
-        | Storage_interface.Storage_error (Backend_error (code, _))
-          when code = "SR_BACKEND_FAILURE_44" ->
-            raise
-              (Api_errors.Server_error
-                 ( Api_errors.sr_source_space_insufficient
-                 , [Storage_interface.Sr.string_of sr]
-                 )
-              )
-        | e ->
-            raise e
+        Storage_smapiv1_migrate.mirror_snapshot ~dbg ~sr ~dp ~mirror_id
+          ~local_vdi
       in
-      SXM.info "%s: snapshot created, mirror initiated vdi:%s snapshot_of:%s"
-        __FUNCTION__
-        (Storage_interface.Vdi.string_of snapshot.vdi)
-        (Storage_interface.Vdi.string_of local_vdi.vdi) ;
-      on_fail := (fun () -> Local.VDI.destroy dbg sr snapshot.vdi) :: !on_fail ;
-      (let rec inner () =
-         let alm_opt = State.find_active_local_mirror mirror_id in
-         match alm_opt with
-         | Some alm ->
-             let stats = Tapctl.stats (Tapctl.create ()) tapdev in
-             if stats.Tapctl.Stats.nbd_mirror_failed = 1 then (
-               error "Tapdisk mirroring has failed" ;
-               Updates.add (Dynamic.Mirror mirror_id) updates
-             ) ;
-             alm.State.Send_state.watchdog <-
-               Some
-                 (Scheduler.one_shot scheduler (Scheduler.Delta 5)
-                    "tapdisk_watchdog" inner
-                 )
-         | None ->
-             ()
-       in
-       inner ()
-      ) ;
-      on_fail := (fun () -> stop ~dbg ~id:mirror_id) :: !on_fail ;
-      (* Copy the snapshot to the remote *)
+      Storage_smapiv1_migrate.mirror_checker mirror_id tapdev ;
       let new_parent =
-        Storage_task.with_subtask task "copy" (fun () ->
-            copy_into_vdi ~task ~dbg ~sr ~vdi:snapshot.vdi ~vm:copy_vm ~url
-              ~dest ~dest_vdi:result.Mirror.copy_diffs_to ~verify_dest
-        )
-        |> vdi_info
+        Storage_smapiv1_migrate.mirror_copy ~task ~dbg ~sr ~snapshot ~copy_vm
+          ~url ~dest_sr:dest ~remote_mirror ~verify_dest
       in
       debug "Local VDI %s = remote VDI %s"
         (Storage_interface.Vdi.string_of snapshot.vdi)
         (Storage_interface.Vdi.string_of new_parent.vdi) ;
       debug "Local VDI %s now mirrored to remote VDI: %s"
         (Storage_interface.Vdi.string_of local_vdi.vdi)
-        (Storage_interface.Vdi.string_of result.Mirror.mirror_vdi.vdi) ;
+        (Storage_interface.Vdi.string_of remote_mirror.Mirror.mirror_vdi.vdi) ;
       debug "Destroying snapshot on src" ;
       Local.VDI.destroy dbg sr snapshot.vdi ;
       Some (Mirror_id mirror_id)
     with
     | Storage_error (Sr_not_attached sr_uuid) ->
         error " Caught exception %s:%s. Performing cleanup."
           Api_errors.sr_not_attached sr_uuid ;
-        perform_cleanup_actions !on_fail ;
         raise (Api_errors.Server_error (Api_errors.sr_not_attached, [sr_uuid]))
+    | ( Storage_error (Migration_mirror_fd_failure reason)
+      | Storage_error (Migration_mirror_snapshot_failure reason) ) as e ->
+        error "%s: Caught %s: during storage migration preparation" __FUNCTION__
+          reason ;
+        MigrateRemote.receive_cancel ~dbg ~id:mirror_id ;
+        raise e
+    | Storage_error (Migration_mirror_copy_failure reason) as e ->
+        error "%s: Caught %s: during storage migration copy" __FUNCTION__ reason ;
+        stop ~dbg ~id:mirror_id ;
+        raise e
     | e ->
-        error "Caught %s: performing cleanup actions" (Api_errors.to_string e) ;
-        perform_cleanup_actions !on_fail ;
+        error "Caught %s during SXM: " (Api_errors.to_string e) ;
+        stop ~dbg ~id:mirror_id ;
         raise e
 
   let stat ~dbg:_ ~id =

ocaml/xapi/storage_smapiv1_migrate.ml

@@ -401,6 +401,161 @@ module Copy = struct
         raise (Storage_error (Internal_error (Printexc.to_string e)))
 end
 
+let mirror_pass_fds ~dbg ~dp ~sr ~vdi ~mirror_vm ~mirror_id ~url ~dest_sr
+    ~verify_dest ~(remote_mirror : Mirror.mirror_receive_result_vhd_t) =
+  let remote_vdi = remote_mirror.mirror_vdi.vdi in
+  let mirror_dp = remote_mirror.mirror_datapath in
+
+  let uri =
+    Printf.sprintf "/services/SM/nbd/%s/%s/%s/%s"
+      (Storage_interface.Vm.string_of mirror_vm)
+      (Storage_interface.Sr.string_of dest_sr)
+      (Storage_interface.Vdi.string_of remote_vdi)
+      mirror_dp
+  in
+  D.debug "%s: uri of http request for mirroring is %s" __FUNCTION__ uri ;
+  let dest_url = Http.Url.set_uri (Http.Url.of_string url) uri in
+  D.debug "%s url of http request for mirroring is %s" __FUNCTION__
+    (Http.Url.to_string dest_url) ;
+  let request =
+    Http.Request.make
+      ~query:(Http.Url.get_query_params dest_url)
+      ~version:"1.0" ~user_agent:"smapiv2" Http.Put uri
+  in
+  let verify_cert = if verify_dest then Stunnel_client.pool () else None in
+  let transport = Xmlrpc_client.transport_of_url ~verify_cert dest_url in
+  D.debug "Searching for data path: %s" dp ;
+  let attach_info = Local.DP.attach_info dbg sr vdi dp mirror_vm in
+
+  let tapdev =
+    match tapdisk_of_attach_info attach_info with
+    | Some tapdev ->
+        let pid = Tapctl.get_tapdisk_pid tapdev in
+        let path = Printf.sprintf "/var/run/blktap-control/nbdclient%d" pid in
+        with_transport ~stunnel_wait_disconnect:false transport
+          (with_http request (fun (_response, s) ->
+               (* Enable mirroring on the local machine *)
+               let control_fd = Unix.socket Unix.PF_UNIX Unix.SOCK_STREAM 0 in
+               finally
+                 (fun () ->
+                   Unix.connect control_fd (Unix.ADDR_UNIX path) ;
+                   let msg = dp in
+                   let len = String.length msg in
+                   let written =
+                     Unixext.send_fd_substring control_fd msg 0 len [] s
+                   in
+                   if written <> len then (
+                     D.error "Failed to transfer fd to %s" path ;
+                     failwith "Internal error transferring fd to tapdisk"
+                   )
+                 )
+                 (fun () -> Unix.close control_fd)
+           )
+          ) ;
+        tapdev
+    | None ->
+        D.error "%s: vdi %s not attached" __FUNCTION__ (Vdi.string_of vdi) ;
+        raise
+          (Storage_interface.Storage_error
+             (Migration_mirror_fd_failure "VDI Not Attached")
+          )
+    | exception e ->
+        D.error "%s Caught exception %s:. Performing cleanup." __FUNCTION__
+          (Printexc.to_string e) ;
+        raise
+          (Storage_interface.Storage_error
+             (Migration_mirror_fd_failure (Printexc.to_string e))
+          )
+  in
+  D.debug "%s Updating active local mirrors: id=%s" __FUNCTION__ mirror_id ;
+  let alm =
+    State.Send_state.
+      {
+        url
+      ; dest_sr
+      ; remote_info=
+          Some
+            {
+              dp= remote_mirror.mirror_datapath
+            ; vdi= remote_mirror.mirror_vdi.vdi
+            ; url
+            ; verify_dest
+            }
+      ; local_dp= dp
+      ; tapdev= Some tapdev
+      ; failed= false
+      ; watchdog= None
+      }
+  in
+  State.add mirror_id (State.Send_op alm) ;
+  D.debug "%s Updated mirror_id %s in the active local mirror" __FUNCTION__
+    mirror_id ;
+  tapdev
+
+let mirror_snapshot ~dbg ~sr ~dp ~mirror_id ~local_vdi =
+  SXM.info "%s About to snapshot VDI = %s" __FUNCTION__
+    (string_of_vdi_info local_vdi) ;
+  let local_vdi = add_to_sm_config local_vdi "mirror" ("nbd:" ^ dp) in
+  let local_vdi = add_to_sm_config local_vdi "base_mirror" mirror_id in
+  let snapshot =
+    try Local.VDI.snapshot dbg sr local_vdi with
+    | Storage_interface.Storage_error (Backend_error (code, _))
+      when code = "SR_BACKEND_FAILURE_44" ->
+        raise
+          (Storage_interface.Storage_error
+             (Migration_mirror_snapshot_failure
+                (Printf.sprintf "%s:%s" Api_errors.sr_source_space_insufficient
+                   (Storage_interface.Sr.string_of sr)
+                )
+             )
+          )
+    | e ->
+        raise
+          (Storage_interface.Storage_error
+             (Migration_mirror_snapshot_failure (Printexc.to_string e))
+          )
+  in
+
+  SXM.info "%s: snapshot created, mirror initiated vdi:%s snapshot_of:%s"
+    __FUNCTION__
+    (Storage_interface.Vdi.string_of snapshot.vdi)
+    (Storage_interface.Vdi.string_of local_vdi.vdi) ;
+
+  snapshot
+
+let mirror_checker mirror_id tapdev =
+  let rec inner () =
+    let alm_opt = State.find_active_local_mirror mirror_id in
+    match alm_opt with
+    | Some alm ->
+        let stats = Tapctl.stats (Tapctl.create ()) tapdev in
+        if stats.Tapctl.Stats.nbd_mirror_failed = 1 then (
+          D.error "Tapdisk mirroring has failed" ;
+          Updates.add (Dynamic.Mirror mirror_id) updates
+        ) ;
+        alm.State.Send_state.watchdog <-
+          Some
+            (Scheduler.one_shot scheduler (Scheduler.Delta 5) "tapdisk_watchdog"
+               inner
+            )
+    | None ->
+        ()
+  in
+  inner ()
+
+let mirror_copy ~task ~dbg ~sr ~snapshot ~copy_vm ~url ~dest_sr ~remote_mirror
+    ~verify_dest =
+  (* Copy the snapshot to the remote *)
+  try
+    Storage_task.with_subtask task "copy" (fun () ->
+        Copy.copy_into_vdi ~task ~dbg ~sr ~vdi:snapshot.vdi ~vm:copy_vm ~url
+          ~dest:dest_sr ~dest_vdi:remote_mirror.Mirror.copy_diffs_to
+          ~verify_dest
+    )
+    |> vdi_info
+  with e ->
+    raise (Storage_error (Migration_mirror_copy_failure (Printexc.to_string e)))
+
 module MIRROR : SMAPIv2_MIRROR = struct
   type context = unit