CP-45016: Implement checking of mirroring features

SXM using qemu datapaths is done in two parts: inbound and outbound,
hence the storage side has declared a `VDI_MIRROR_IN` features.
Implement the checking logic for this feature here. This is done in two
parts:

1. Check the relevant SM for the feature `VDI_MIRROR_IN`. This is done
   on the destination host. If this fails, then storage migration will
   be prevented from happening in the beginning. Consider this as a
   pre-check.
2. Check in xapi-storage-script that the chosen datapath supports
   mirroring. Technically mirroring is a datapath feature so it makes
   sense to check it there. But currently there is no easy way to
   register a datapath feature into the SM object in xapi database, so
   instead the volume plugins for SMAPIv3 SRs would also declare this
   feature. However, we still need a final check on the datapath
   features to make sure the selected datapath really supports this
   feature. Note this check happens while trying to set up the mirror,
   after storage migration has started, and will give a relatively
   cryptic error to the user.

Signed-off-by: Vincent Liu <[email protected]>

ocaml/xapi-storage-script/main.ml

@@ -335,6 +335,8 @@ end
 
 let _nonpersistent = "NONPERSISTENT"
 
+let _vdi_mirror_in = "VDI_MIRROR_IN"
+
 let _clone_on_boot_key = "clone-on-boot"
 
 let _vdi_type_key = "vdi-type"
@@ -1750,13 +1752,13 @@ let bind ~volume_script_dir =
          stat ~dbg ~sr ~vdi:temporary
      )
      >>>= fun response ->
-     choose_datapath domain response >>>= fun (rpc, _datapath, uri, domain) ->
-     return_data_rpc (fun () ->
-         (* debug
-            "about to call Datapath_client.import_activate with uri %s domain %s"
-            uri domain ; *)
-         Datapath_client.import_activate (rpc ~dbg) dbg uri domain
-     )
+     choose_datapath domain response >>>= fun (rpc, datapath, uri, domain) ->
+     if Datapath_plugins.supports_feature datapath _vdi_mirror_in then
+       return_data_rpc (fun () ->
+           Datapath_client.import_activate (rpc ~dbg) dbg uri domain
+       )
+     else
+       fail (Storage_interface.Errors.Unimplemented _vdi_mirror_in)
     )
     |> wrap
   in

ocaml/xapi/smint.ml

@@ -41,6 +41,7 @@ module Feature = struct
     | Vdi_attach
     | Vdi_detach
     | Vdi_mirror
+    | Vdi_mirror_in
     | Vdi_clone
     | Vdi_snapshot
     | Vdi_resize
@@ -79,6 +80,7 @@ module Feature = struct
     ; ("VDI_ATTACH", Vdi_attach)
     ; ("VDI_DETACH", Vdi_detach)
     ; ("VDI_MIRROR", Vdi_mirror)
+    ; ("VDI_MIRROR_IN", Vdi_mirror_in)
     ; ("VDI_RESIZE", Vdi_resize)
     ; ("VDI_RESIZE_ONLINE", Vdi_resize_online)
     ; ("VDI_CLONE", Vdi_clone)
@@ -134,7 +136,17 @@ module Feature = struct
   in the feature list [fl]. Callers should use this function to test if a feature
     is available rather than directly using membership functions on a feature list
     as this function might have special logic for some features. *)
-  let has_capability (c : capability) fl = List.mem_assoc c fl
+  let has_capability (c : capability) (fl : t list) =
+    List.fold_left
+      (fun acc fe ->
+        match fe with
+        | Vdi_mirror, n ->
+            (Vdi_mirror_in, n) :: (Vdi_mirror, n) :: acc
+        | f ->
+            f :: acc
+      )
+      [] fl
+    |> List.mem_assoc c
 
   (** [parse_string_int64 features] takes a [features] list in its plain string
   forms such as "VDI_MIRROR/2" and parses them into the form of (VDI_MIRROR, 2).

ocaml/xapi/storage_migrate.ml

@@ -1234,7 +1234,11 @@ let nbd_handler req s ?(vm = "0") sr vdi dp =
   let sr, vdi = Storage_interface.(Sr.of_string sr, Vdi.of_string vdi) in
   req.Http.Request.close <- true ;
   let vm = vm_of_s vm in
-  let path = Local.DATA.MIRROR.get_nbd_server "nbd" dp sr vdi vm in
+  let path =
+    Storage_utils.transform_storage_exn (fun () ->
+        Local.DATA.MIRROR.get_nbd_server "nbd" dp sr vdi vm
+    )
+  in
   Http_svr.headers s (Http.http_200_ok () @ ["Transfer-encoding: nbd"]) ;
   let control_fd = Unix.socket Unix.PF_UNIX Unix.SOCK_STREAM 0 in
   finally

ocaml/xapi/xapi_vm_migrate.ml

@@ -160,7 +160,8 @@ end))
 
 open Storage_interface
 
-let assert_sr_support_operations ~__context ~vdi_map ~remote ~ops =
+let assert_sr_support_operations ~__context ~vdi_map ~remote ~local_ops
+    ~remote_ops =
   let op_supported_on_source_sr vdi ops =
     let open Smint.Feature in
     (* Check VDIs must not be present on SR which doesn't have required capability *)
@@ -211,8 +212,8 @@ let assert_sr_support_operations ~__context ~vdi_map ~remote ~ops =
   in
   List.filter (fun (vdi, sr) -> not (is_sr_matching vdi sr)) vdi_map
   |> List.iter (fun (vdi, sr) ->
-         op_supported_on_source_sr vdi ops ;
-         op_supported_on_dest_sr sr ops sm_record remote
+         op_supported_on_source_sr vdi local_ops ;
+         op_supported_on_dest_sr sr remote_ops sm_record remote
      )
 
 (** Check that none of the VDIs that are mapped to a different SR have CBT
@@ -1771,8 +1772,9 @@ let assert_can_migrate ~__context ~vm ~dest ~live:_ ~vdi_map ~vif_map ~options
     )
     vms_vdis ;
   (* operations required for migration *)
-  let required_sr_operations =
-    [Smint.Feature.Vdi_mirror; Smint.Feature.Vdi_snapshot]
+  let required_src_sr_operations = Smint.Feature.[Vdi_snapshot; Vdi_mirror] in
+  let required_dst_sr_operations =
+    Smint.Feature.[Vdi_snapshot; Vdi_mirror_in]
   in
   let host_from = Helpers.LocalObject source_host_ref in
   ( match migration_type ~__context ~remote with
@@ -1786,7 +1788,8 @@ let assert_can_migrate ~__context ~vm ~dest ~live:_ ~vdi_map ~vif_map ~options
           (Api_errors.Server_error (Api_errors.not_supported_during_upgrade, [])) ;
       (* Check VDIs are not migrating to or from an SR which doesn't have required_sr_operations *)
       assert_sr_support_operations ~__context ~vdi_map ~remote
-        ~ops:required_sr_operations ;
+        ~local_ops:required_src_sr_operations
+        ~remote_ops:required_dst_sr_operations ;
       let snapshot = Db.VM.get_record ~__context ~self:vm in
       let do_cpuid_check = not force in
       Xapi_vm_helpers.assert_can_boot_here ~__context ~self:vm
@@ -1818,7 +1821,8 @@ let assert_can_migrate ~__context ~vm ~dest ~live:_ ~vdi_map ~vif_map ~options
       let power_state = Db.VM.get_power_state ~__context ~self:vm in
       (* Check VDIs are not migrating to or from an SR which doesn't have required_sr_operations *)
       assert_sr_support_operations ~__context ~vdi_map ~remote
-        ~ops:required_sr_operations ;
+        ~local_ops:required_src_sr_operations
+        ~remote_ops:required_dst_sr_operations ;
       (* The copy mode is only allow on stopped VM *)
       if (not force) && copy && power_state <> `Halted then
         raise