Merge pull request #35 from djs55/forupstream/CP-12902

Register and deregister datasources with xcp-rrdd

Author: djs55

_oasis

@@ -13,4 +13,4 @@ Executable xapi_script_storage
   MainIs:             main.ml
   Custom:             true
   Install:            false
-  BuildDepends:       xcp, xcp.storage, async_inotify, threads, message_switch.async (>= 0.11.0), rpclib, xapi-storage, sexplib, sexplib.syntax, rpclib, rpclib.syntax
+  BuildDepends:       xcp, xcp.storage, xcp.rrd, async_inotify, threads, message_switch.async (>= 0.11.0), rpclib, xapi-storage, sexplib, sexplib.syntax, rpclib, rpclib.syntax

_tags

@@ -1,5 +1,5 @@
 # OASIS_START
-# DO NOT EDIT (digest: d4caced815577ed0a5b4b8487e4f2276)
+# DO NOT EDIT (digest: 2c22384c8337719c5d75ab0e5fc90afb)
 # Ignore VCS directories, you can use the same kind of rule outside
 # OASIS_START/STOP if you want to exclude directories that contains
 # useless stuff for the build process
@@ -24,6 +24,7 @@ true: annot, bin_annot
 <main.{native,byte}>: pkg_threads
 <main.{native,byte}>: pkg_xapi-storage
 <main.{native,byte}>: pkg_xcp
+<main.{native,byte}>: pkg_xcp.rrd
 <main.{native,byte}>: pkg_xcp.storage
 <*.ml{,i,y}>: pkg_async_inotify
 <*.ml{,i,y}>: pkg_message_switch.async
@@ -34,6 +35,7 @@ true: annot, bin_annot
 <*.ml{,i,y}>: pkg_threads
 <*.ml{,i,y}>: pkg_xapi-storage
 <*.ml{,i,y}>: pkg_xcp
+<*.ml{,i,y}>: pkg_xcp.rrd
 <*.ml{,i,y}>: pkg_xcp.storage
 <main.{native,byte}>: custom
 # OASIS_STOP

main.ml

@@ -39,6 +39,35 @@ let info  fmt = log Core.Syslog.Level.INFO    fmt
 let warn  fmt = log Core.Syslog.Level.WARNING fmt
 let error fmt = log Core.Syslog.Level.ERR     fmt
 
+module RRD = struct
+  open Protocol_async
+
+  let (>>|=) m f = m >>= function
+    | `Ok x -> f x
+    | `Error y ->
+      let b = Buffer.create 16 in
+      let fmt = Format.formatter_of_buffer b in
+      Client.pp_error fmt y;
+      Format.pp_print_flush fmt ();
+      raise (Failure (Buffer.contents b))
+
+  let switch_rpc queue_name string_of_call response_of_string call =
+    Client.connect ~switch:queue_name () >>|= fun t ->
+    Client.rpc ~t ~queue:queue_name ~body:(string_of_call call) () >>|= fun s ->
+    return (response_of_string s)
+
+  let json_switch_rpc queue_name = switch_rpc queue_name Jsonrpc.string_of_call Jsonrpc.response_of_string
+
+  module Client = Rrd_interface.ClientM(struct
+    type 'a t = 'a Deferred.t
+    let return = return
+    let bind = Deferred.bind
+    let fail = raise
+    let rpc call = json_switch_rpc !Rrd_interface.queue_name call
+end)
+
+end
+
 let _nonpersistent = "NONPERSISTENT"
 let _clone_on_boot_key = "clone-on-boot"
 
@@ -129,16 +158,21 @@ let script root_dir name kind script = match kind with
 | `Datapath datapath -> Filename.(concat (concat (concat (dirname root_dir) "datapath") datapath) script)
 
 module Attached_SRs = struct
-  let sr_table : string String.Table.t ref = ref (String.Table.create ())
+  type state = {
+    sr: string;
+    uids: string list;
+  } with sexp
+
+  let sr_table : state String.Table.t ref = ref (String.Table.create ())
   let state_path = ref None
 
-  let add smapiv2 plugin =
-    Hashtbl.replace !sr_table smapiv2 plugin;
+  let add smapiv2 plugin uids =
+    Hashtbl.replace !sr_table smapiv2 { sr = plugin; uids };
     ( match !state_path with
       | None ->
         return ()
       | Some path ->
-        let contents = String.Table.sexp_of_t (fun x -> Sexplib.Sexp.Atom x) !sr_table |> Sexplib.Sexp.to_string in
+        let contents = String.Table.sexp_of_t sexp_of_state !sr_table |> Sexplib.Sexp.to_string in
         Writer.save path ~contents
     ) >>= fun () ->
     return (Ok ())
@@ -149,7 +183,15 @@ module Attached_SRs = struct
       let open Storage_interface in
       let exnty = Exception.Sr_not_attached smapiv2 in
       return (Error (Exception.rpc_of_exnty exnty))
-    | Some sr -> return (Ok sr)
+    | Some { sr } -> return (Ok sr)
+
+  let get_uids smapiv2 =
+    match Hashtbl.find !sr_table smapiv2 with
+    | None ->
+      let open Storage_interface in
+      let exnty = Exception.Sr_not_attached smapiv2 in
+      return (Error (Exception.rpc_of_exnty exnty))
+    | Some { uids } -> return (Ok uids)
 
   let remove smapiv2 =
     Hashtbl.remove !sr_table smapiv2;
@@ -164,7 +206,7 @@ module Attached_SRs = struct
     | `Yes ->
       Reader.file_contents path
       >>= fun contents ->
-      sr_table := contents |> Sexplib.Sexp.of_string |> String.Table.t_of_sexp (function Sexplib.Sexp.Atom x -> x | _ -> assert false);
+      sr_table := contents |> Sexplib.Sexp.of_string |> String.Table.t_of_sexp state_of_sexp;
       return ()
 end
 
@@ -362,11 +404,36 @@ let process root_dir name x =
       let args' = Storage.Volume.Types.SR.Attach.In.rpc_of_t args' in
       let open Deferred.Result.Monad_infix in
       fork_exec_rpc root_dir (script root_dir name `Volume "SR.attach") args' Storage.Volume.Types.SR.Attach.Out.t_of_rpc
-      >>= fun response ->
+      >>= fun attach_response ->
+      let sr = args.Args.SR.Attach.sr in
+      (* Stat the SR to look for datasources *)
+      let args = Storage.Volume.Types.SR.Stat.In.make
+        args.Args.SR.Attach.dbg
+        uri in
+      let args = Storage.Volume.Types.SR.Stat.In.rpc_of_t args in
+      fork_exec_rpc root_dir (script root_dir name `Volume "SR.stat") args Storage.Volume.Types.SR.Stat.Out.t_of_rpc
+      >>= fun stat ->
+      let open Deferred.Monad_infix in
+      let rec loop acc = function
+      | [] -> return acc
+      | datasource :: datasources ->
+        let uri = Uri.of_string datasource in
+        match Uri.scheme uri with
+        | Some "xeno+shm" ->
+          let uid = Uri.path uri in
+          let uid = if String.length uid > 1 then String.sub uid 1 (String.length uid - 1) else uid in
+          RRD.Client.Plugin.Local.register ~uid ~info:Rrd.Five_Seconds ~protocol:Rrd_interface.V2
+          >>= fun _ ->
+          loop (uid :: acc) datasources
+        | _ ->
+          loop acc datasources in
+      loop [] stat.Storage.Volume.Types.datasources
+      >>= fun uids ->
+      let open Deferred.Result.Monad_infix in
       (* associate the 'sr' from the plugin with the SR reference passed in *)
-      Attached_SRs.add args.Args.SR.Attach.sr response
+      Attached_SRs.add sr attach_response uids
       >>= fun () ->
-      Deferred.Result.return (R.success (Args.SR.Attach.rpc_of_response response))
+      Deferred.Result.return (R.success (Args.SR.Attach.rpc_of_response attach_response))
     end
   | { R.name = "SR.detach"; R.params = [ args ] } ->
     let args = Args.SR.Detach.request_of_rpc args in
@@ -383,6 +450,25 @@ let process root_dir name x =
       let args' = Storage.Volume.Types.SR.Detach.In.rpc_of_t args' in
       fork_exec_rpc root_dir (script root_dir name `Volume "SR.detach") args' Storage.Volume.Types.SR.Detach.Out.t_of_rpc
       >>= fun response ->
+      Attached_SRs.get_uids args.Args.SR.Detach.sr
+      >>= fun uids ->
+      let open Deferred.Monad_infix in
+      let rec loop = function
+      | [] -> return ()
+      | datasource :: datasources ->
+        let uri = Uri.of_string datasource in
+        match Uri.scheme uri with
+        | Some "xeno+shm" ->
+          let uid = Uri.path uri in
+          let uid = if String.length uid > 1 then String.sub uid 1 (String.length uid - 1) else uid in
+          RRD.Client.Plugin.Local.deregister ~uid
+          >>= fun _ ->
+          loop datasources
+        | _ ->
+          loop datasources in
+      loop uids
+      >>= fun () ->
+      let open Deferred.Result.Monad_infix in
       Attached_SRs.remove args.Args.SR.Detach.sr
       >>= fun () ->
       Deferred.Result.return (R.success (Args.SR.Detach.rpc_of_response response))

setup.ml

@@ -1,7 +1,7 @@
 (* setup.ml generated for the first time by OASIS v0.4.4 *)
 
 (* OASIS_START *)
-(* DO NOT EDIT (digest: 29609653515f2db19f5b74fca28cd6eb) *)
+(* DO NOT EDIT (digest: 7cbbdca01feb98c65e73ab10fd15a812) *)
 (*
    Regenerated by OASIS v0.4.5
    Visit http://oasis.forge.ocamlcore.org for more information and
@@ -6741,6 +6741,7 @@ let setup_t =
                         [
                            FindlibPackage ("xcp", None);
                            FindlibPackage ("xcp.storage", None);
+                           FindlibPackage ("xcp.rrd", None);
                            FindlibPackage ("async_inotify", None);
                            FindlibPackage ("threads", None);
                            FindlibPackage
@@ -6772,14 +6773,14 @@ let setup_t =
        };
      oasis_fn = Some "_oasis";
      oasis_version = "0.4.5";
-     oasis_digest = Some "\012@\028\131\159\175G\238\214P\150/AW\151F";
+     oasis_digest = Some "\224\128\205S\160\162\"\236\\\2334g_\202\162\133";
      oasis_exec = None;
      oasis_setup_args = [];
      setup_update = false
   };;
 
 let setup () = BaseSetup.setup setup_t;;
 
-# 6784 "setup.ml"
+# 6785 "setup.ml"
 (* OASIS_STOP *)
 let () = setup ();;