CA-409431: Use an Observer forwarder for xapi-storage-script

Currently, xapi-storage-script uses the presence/absence of a smapi
observer config file to determine whether it should create traces. This
only happens on startup which means smapiv3 traces will often not be
created when they should be.

This commit updates the Smapi Observer forwarder to use an RPC client to
send messages to xapi-storage-script, updating it on any relevant
changes to the Observer.

Author: snwoods

ocaml/xapi-idl/lib/dune

@@ -40,7 +40,7 @@
  (wrapped false)
  (preprocess
    (per_module
-     ((pps ppx_deriving_rpc) Xcp_channel Xcp_channel_protocol TypeCombinators Observer_helpers)
+     ((pps ppx_deriving_rpc) Xcp_channel Xcp_channel_protocol TypeCombinators Observer_helpers Observer_skeleton)
      ((pps ppx_sexp_conv ppx_deriving_rpc) Xcp_pci))))
 
 (library

ocaml/xapi-idl/lib/observer_helpers.ml

@@ -19,6 +19,16 @@ module D = Debug.Make (struct let name = "observer_interface" end)
 
 open D
 
+let service_name = "observer"
+
+let queue_name = Xcp_service.common_prefix ^ service_name
+
+let default_sockets_dir = "/var/lib/xcp"
+
+let default_path = Filename.concat default_sockets_dir service_name
+
+let uri () = "file:" ^ default_path
+
 module Errors = struct
   type error =
     | Internal_error of string
@@ -148,3 +158,107 @@ module ObserverAPI (R : RPC) = struct
     declare "Observer.set_compress_tracing_files" []
       (dbg_p @-> bool_p @-> returning unit_p err)
 end
+
+module type Server_impl = sig
+  type context = unit
+
+  val create :
+       context
+    -> dbg:debug_info
+    -> uuid:string
+    -> name_label:string
+    -> attributes:(string * string) list
+    -> endpoints:string list
+    -> enabled:bool
+    -> unit
+
+  val destroy : context -> dbg:debug_info -> uuid:string -> unit
+
+  val set_enabled :
+    context -> dbg:debug_info -> uuid:string -> enabled:bool -> unit
+
+  val set_attributes :
+       context
+    -> dbg:debug_info
+    -> uuid:string
+    -> attributes:(string * string) list
+    -> unit
+
+  val set_endpoints :
+    context -> dbg:debug_info -> uuid:string -> endpoints:string list -> unit
+
+  val init : context -> dbg:debug_info -> unit
+
+  val set_trace_log_dir : context -> dbg:debug_info -> dir:string -> unit
+
+  val set_export_interval : context -> dbg:debug_info -> interval:float -> unit
+
+  val set_max_spans : context -> dbg:debug_info -> spans:int -> unit
+
+  val set_max_traces : context -> dbg:debug_info -> traces:int -> unit
+
+  val set_max_file_size : context -> dbg:debug_info -> file_size:int -> unit
+
+  val set_host_id : context -> dbg:debug_info -> host_id:string -> unit
+
+  val set_compress_tracing_files :
+    context -> dbg:debug_info -> enabled:bool -> unit
+end
+
+module Server (Impl : Server_impl) () = struct
+  module S = ObserverAPI (Idl.Exn.GenServer ())
+
+  let _ =
+    S.create (fun dbg uuid name_label attributes endpoints enabled ->
+        Impl.create () ~dbg ~uuid ~name_label ~attributes ~endpoints ~enabled
+    ) ;
+    S.destroy (fun dbg uuid -> Impl.destroy () ~dbg ~uuid) ;
+    S.set_enabled (fun dbg uuid enabled ->
+        Impl.set_enabled () ~dbg ~uuid ~enabled
+    ) ;
+    S.set_attributes (fun dbg uuid attributes ->
+        Impl.set_attributes () ~dbg ~uuid ~attributes
+    ) ;
+    S.set_endpoints (fun dbg uuid endpoints ->
+        Impl.set_endpoints () ~dbg ~uuid ~endpoints
+    ) ;
+    S.init (fun dbg -> Impl.init () ~dbg) ;
+    S.set_trace_log_dir (fun dbg dir -> Impl.set_trace_log_dir () ~dbg ~dir) ;
+    S.set_export_interval (fun dbg interval ->
+        Impl.set_export_interval () ~dbg ~interval
+    ) ;
+    S.set_max_spans (fun dbg spans -> Impl.set_max_spans () ~dbg ~spans) ;
+    S.set_max_traces (fun dbg traces -> Impl.set_max_traces () ~dbg ~traces) ;
+    S.set_max_file_size (fun dbg file_size ->
+        Impl.set_max_file_size () ~dbg ~file_size
+    ) ;
+    S.set_host_id (fun dbg host_id -> Impl.set_host_id () ~dbg ~host_id) ;
+    S.set_compress_tracing_files (fun dbg enabled ->
+        Impl.set_compress_tracing_files () ~dbg ~enabled
+    )
+
+  (* Bind all *)
+  let process call = Idl.Exn.server S.implementation call
+end
+
+let rec retry_econnrefused f =
+  try f () with
+  | Unix.Unix_error (Unix.ECONNREFUSED, "connect", _) ->
+      (* debug "Caught ECONNREFUSED; retrying in 5s"; *)
+      Thread.delay 5. ; retry_econnrefused f
+  | e ->
+      (* error "Caught %s: does the observer service need restarting?"
+         (Printexc.to_string e); *)
+      raise e
+
+module Client = ObserverAPI (Idl.Exn.GenClient (struct
+  open Xcp_client
+
+  let rpc call =
+    retry_econnrefused (fun () ->
+        if !use_switch then
+          json_switch_rpc queue_name call
+        else
+          xml_http_rpc ~srcstr:(get_user_agent ()) ~dststr:queue_name uri call
+    )
+end))

ocaml/xapi-idl/lib/observer_skeleton.ml

@@ -0,0 +1,48 @@
+(*
+ * Copyright (c) Cloud Software Group
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *)
+[@@@ocaml.warning "-27"]
+
+let u x = raise Observer_helpers.(Observer_error (Errors.Unimplemented x))
+
+module Observer = struct
+  type context = unit
+
+  let create ctx ~dbg ~uuid ~name_label ~attributes ~endpoints ~enabled =
+    u "Observer.create"
+
+  let destroy ctx ~dbg ~uuid = u "Observer.destroy"
+
+  let set_enabled ctx ~dbg ~uuid ~enabled = u "Observer.set_enabled"
+
+  let set_attributes ctx ~dbg ~uuid ~attributes = u "Observer.set_attributes"
+
+  let set_endpoints ctx ~dbg ~uuid ~endpoints = u "Observer.set_endpoints"
+
+  let init ctx ~dbg = u "Observer.init"
+
+  let set_trace_log_dir ctx ~dbg ~dir = u "Observer.set_trace_log_dir"
+
+  let set_export_interval ctx ~dbg ~interval = u "Observer.set_export_interval"
+
+  let set_max_spans ctx ~dbg ~spans = u "Observer.set_max_spans"
+
+  let set_max_traces ctx ~dbg ~traces = u "Observer.set_max_traces"
+
+  let set_max_file_size ctx ~dbg ~file_size = u "Observer.set_max_file_size"
+
+  let set_host_id ctx ~dbg ~host_id = u "Observer.set_host_id"
+
+  let set_compress_tracing_files ctx ~dbg ~enabled =
+    u "Observer.set_compress_tracing_files"
+end

ocaml/xapi-storage-script/dune

@@ -41,6 +41,7 @@
     sexplib
     sexplib0
     uri
+    threads.posix
     xapi-backtrace
     xapi-consts
     xapi-consts.xapi_version

ocaml/xapi-storage-script/main.ml

@@ -411,19 +411,6 @@ let observer_config_dir =
   in
   dir // component // "enabled"
 
-(** Determine if SM API observation is enabled from the
-    filesystem. Ordinarily, determining if a component is enabled
-    would consist of querying the 'components' field of an observer
-    from the xapi database. *)
-let observer_is_component_enabled () =
-  let is_enabled () =
-    let is_config_file path = Filename.check_suffix path ".observer.conf" in
-    let* files = Sys.readdir observer_config_dir in
-    Lwt.return (List.exists is_config_file files)
-  in
-  let* result = Deferred.try_with is_enabled in
-  Lwt.return (Option.value (Result.to_option result) ~default:false)
-
 (** Call the script named after the RPC method in the [script_dir]
     directory. The arguments (not the whole JSON-RPC call) are passed as JSON
     to its stdin, and stdout is returned. In case of a non-zero exit code,
@@ -2247,6 +2234,19 @@ let register_exn_pretty_printers () =
         assert false
     )
 
+module XapiStorageScript : Observer_helpers.Server_impl = struct
+  include Observer_skeleton.Observer
+
+  let create _context ~dbg:_ ~uuid:_ ~name_label:_ ~attributes:_ ~endpoints:_
+      ~enabled =
+    config.use_observer <- enabled
+
+  let destroy _context ~dbg:_ ~uuid:_ = config.use_observer <- false
+
+  let set_enabled _context ~dbg:_ ~uuid:_ ~enabled =
+    config.use_observer <- enabled
+end
+
 let () =
   register_exn_pretty_printers () ;
   let root_dir = ref "/var/lib/xapi/storage-scripts" in
@@ -2293,9 +2293,17 @@ let () =
 
   Logs.set_reporter (lwt_reporter ()) ;
   Logs.set_level ~all:true (Some Logs.Info) ;
+
+  let module S = Observer_helpers.Server (XapiStorageScript) () in
+  let s =
+    Xcp_service.make ~path:Observer_helpers.default_path
+      ~queue_name:Observer_helpers.queue_name ~rpc_fn:S.process ()
+  in
+  let (_ : Thread.t) =
+    Thread.create (fun () -> Xcp_service.serve_forever s) ()
+  in
+
   let main =
-    let* observer_enabled = observer_is_component_enabled () in
-    config.use_observer <- observer_enabled ;
     if !self_test_only then
       self_test ~root_dir:!root_dir
     else

ocaml/xapi/xapi_observer.ml

@@ -383,6 +383,30 @@ end
 
 module SMObserverConfig = Dom0ObserverConfig (struct let component = SMApi end)
 
+module SMObserver = struct
+  include SMObserverConfig
+  open Observer_helpers
+
+  let create ~__context ~uuid ~name_label ~attributes ~endpoints ~enabled =
+    debug "SMObserver Observer.create %s" uuid ;
+    SMObserverConfig.create ~__context ~uuid ~name_label ~attributes ~endpoints
+      ~enabled ;
+    let dbg = Context.string_of_task __context in
+    Client.create dbg uuid name_label attributes endpoints enabled
+
+  let destroy ~__context ~uuid =
+    debug "SMObserver Observer.destroy %s" uuid ;
+    SMObserverConfig.destroy ~__context ~uuid ;
+    let dbg = Context.string_of_task __context in
+    Client.destroy dbg uuid
+
+  let set_enabled ~__context ~uuid ~enabled =
+    debug "SMObserver Observer.set_enabled %s" uuid ;
+    SMObserverConfig.set_enabled ~__context ~uuid ~enabled ;
+    let dbg = Context.string_of_task __context in
+    Client.set_enabled dbg uuid enabled
+end
+
 let get_forwarder c =
   let module Forwarder =
     ( val match c with
@@ -393,7 +417,7 @@ let get_forwarder c =
           | Xapi_clusterd ->
               (module Xapi_cluster.Observer)
           | SMApi ->
-              (module SMObserverConfig)
+              (module SMObserver)
         : ObserverInterface
       )
   in