CP-47063: Tracing instrumentation around message-switch send (#6503)

Instrument xenops vm non-atomic functions.
Instruments:
- `VM.add`,
- `VM.stat`,
- `VM.exists`,
- `VM.list`.

Instruments `switch_rpc` according to OpenTelemetry standard on
instrumenting rpc calls.
- `server.address` is the name of the message queue.

Intruments sending the message on a queue according to OpenTelemetry
standard on instrumenting messaging.
- `destination` is the name of the message queue.

`Tracing.with_tracing` now accepts an optional argument to set the Span
Kind.

![image](https://github.com/user-attachments/assets/b58a3309-6e31-4e40-84a5-c053bc00d5c8)

Author: edwintorok

ocaml/libs/tracing/tracing.ml

@@ -792,10 +792,14 @@ end
 let enable_span_garbage_collector ?(timeout = 86400.) () =
   Spans.GC.initialise_thread ~timeout
 
-let with_tracing ?(attributes = []) ?(parent = None) ?trace_context ~name f =
+let with_tracing ?(attributes = []) ?(parent = None) ?span_kind ?trace_context
+    ~name f =
   let tracer = Tracer.get_tracer ~name in
   if tracer.enabled then (
-    match Tracer.start ~tracer ?trace_context ~attributes ~name ~parent () with
+    match
+      Tracer.start ?span_kind ~tracer ?trace_context ~attributes ~name ~parent
+        ()
+    with
     | Ok span -> (
       try
         let result = f span in

ocaml/libs/tracing/tracing.mli

@@ -190,12 +190,12 @@ module Tracer : sig
     -> (Span.t option, exn) result
 
   val update_span_with_parent : Span.t -> Span.t option -> Span.t option
-  (**[update_span_with_parent s p] returns [Some span] where [span] is an 
+  (**[update_span_with_parent s p] returns [Some span] where [span] is an
   updated verison of the span [s].
-  If [p] is [Some parent], [span] is a child of [parent], otherwise it is the 
+  If [p] is [Some parent], [span] is a child of [parent], otherwise it is the
   original [s].
-  
-  If the span [s] is finished or is no longer considered an on-going span, 
+
+  If the span [s] is finished or is no longer considered an on-going span,
   returns [None].
   *)
 
@@ -209,7 +209,7 @@ module Tracer : sig
   val finished_span_hashtbl_is_empty : unit -> bool
 end
 
-(** [TracerProvider] module provides ways to intereact with the tracer providers. 
+(** [TracerProvider] module provides ways to intereact with the tracer providers.
     *)
 module TracerProvider : sig
   (** Type that represents a tracer provider.*)
@@ -222,7 +222,7 @@ module TracerProvider : sig
     -> name_label:string
     -> uuid:string
     -> unit
-  (** [create ~enabled ~attributes ~endpoints ~name_label ~uuid] initializes a 
+  (** [create ~enabled ~attributes ~endpoints ~name_label ~uuid] initializes a
       tracer provider based on the following parameters: [enabled], [attributes],
       [endpoints], [name_label], and [uuid]. *)
 
@@ -234,17 +234,17 @@ module TracerProvider : sig
     -> unit
     -> unit
   (** [set ?enabled ?attributes ?endpoints ~uuid ()] updates the tracer provider
-      identified by the given [uuid] with the new configuration paremeters: 
-      [enabled], [attributes], and [endpoints]. 
-        
+      identified by the given [uuid] with the new configuration paremeters:
+      [enabled], [attributes], and [endpoints].
+
       If any of the configuration parameters are
       missing, the old ones are kept.
-        
+
       Raises [Failure] if there are no tracer provider with the given [uuid].
       *)
 
   val destroy : uuid:string -> unit
-  (** [destroy ~uuid] destroys the tracer provider with the given [uuid]. 
+  (** [destroy ~uuid] destroys the tracer provider with the given [uuid].
       If there are no tracer provider with the given [uuid], it does nothing.
       *)
 
@@ -269,6 +269,7 @@ val enable_span_garbage_collector : ?timeout:float -> unit -> unit
 val with_tracing :
      ?attributes:(string * string) list
   -> ?parent:Span.t option
+  -> ?span_kind:SpanKind.t
   -> ?trace_context:TraceContext.t
   -> name:string
   -> (Span.t option -> 'a)
@@ -288,24 +289,24 @@ val get_observe : unit -> bool
 
 val validate_attribute : string * string -> bool
 
-(** [EnvHelpers] module is a helper module for the tracing library to easily 
-    transition back and forth between a string list of environment variables to 
-    a traceparent. 
+(** [EnvHelpers] module is a helper module for the tracing library to easily
+    transition back and forth between a string list of environment variables to
+    a traceparent.
     *)
 module EnvHelpers : sig
   val traceparent_key : string
   (** [traceparent_key] is a constant the represents the key of the traceparent
-      environment variable. 
+      environment variable.
       *)
 
   val of_traceparent : string option -> string list
   (** [of_traceparent traceparent_opt] returns a singleton list consisting of a
-      envirentment variable with the key [traceparent_key] and value [v] if 
+      envirentment variable with the key [traceparent_key] and value [v] if
       [traceparent_opt] is [Some v]. Otherwise, returns an empty list. *)
 
   val to_traceparent : string list -> string option
-  (** [to_traceparent env_var_lst] returns [Some v] where v is the value of the 
-      environmental variable coresponding to the key [traceparent_key] from a 
+  (** [to_traceparent env_var_lst] returns [Some v] where v is the value of the
+      environmental variable coresponding to the key [traceparent_key] from a
       string list of environmental variables [env_var_lst]. If there is no such
       evironmental variable in the list, it returns [None].
       *)
@@ -314,7 +315,7 @@ module EnvHelpers : sig
   (** [of_span span] returns a singleton list consisting of a
       envirentment variable with the key [traceparent_key] and value [v], where
       [v] is traceparent representation of span [s] (if [span] is [Some s]).
-        
+
       If [span] is [None], it returns an empty list.
       *)
 end

ocaml/libs/tracing/tracing_export.ml

@@ -278,8 +278,8 @@ module Destination = struct
             ]
           in
           let@ _ =
-            with_tracing ~trace_context:TraceContext.empty ~parent ~attributes
-              ~name
+            with_tracing ~span_kind:Server ~trace_context:TraceContext.empty
+              ~parent ~attributes ~name
           in
           all_spans
           |> Content.Json.ZipkinV2.content_of
@@ -293,8 +293,8 @@ module Destination = struct
     let ((_span_list, span_count) as span_info) = Spans.since () in
     let attributes = [("export.traces.count", string_of_int span_count)] in
     let@ parent =
-      with_tracing ~trace_context:TraceContext.empty ~parent:None ~attributes
-        ~name:"Tracing.flush_spans"
+      with_tracing ~span_kind:Server ~trace_context:TraceContext.empty
+        ~parent:None ~attributes ~name:"Tracing.flush_spans"
     in
     TracerProvider.get_tracer_providers ()
     |> List.filter TracerProvider.get_enabled

ocaml/message-switch/core/dune

@@ -9,6 +9,7 @@
     sexplib
     sexplib0
     threads.posix
+    tracing
     uri
     xapi-log
     xapi-stdext-threads

ocaml/message-switch/core/make.ml

@@ -229,7 +229,7 @@ functor
       in
       return (Ok t)
 
-    let rpc ~t ~queue ?timeout ~body:x () =
+    let rpc ?_span_parent ~t ~queue ?timeout ~body:x () =
       let ivar = M.Ivar.create () in
       let timer =
         Option.map

ocaml/message-switch/core/s.ml

@@ -144,7 +144,8 @@ module type CLIENT = sig
   (** [disconnect] closes the connection *)
 
   val rpc :
-       t:t
+       ?_span_parent:Tracing.Span.t
+    -> t:t
     -> queue:string
     -> ?timeout:int
     -> body:string

ocaml/message-switch/unix/dune

@@ -11,6 +11,7 @@
     rpclib.core
     rpclib.json
     threads.posix
+    tracing
     xapi-stdext-threads
     xapi-stdext-unix
   )

ocaml/message-switch/unix/protocol_unix.ml

@@ -347,7 +347,7 @@ module Client = struct
               Ok c'
       )
 
-  let rpc ~t:c ~queue:dest_queue_name ?timeout ~body:x () =
+  let rpc ?_span_parent ~t:c ~queue:dest_queue_name ?timeout ~body:x () =
     let t = Ivar.create () in
     let timer =
       Option.map
@@ -364,9 +364,23 @@ module Client = struct
             do_rpc c.requests_conn (In.CreatePersistent dest_queue_name)
             >>|= fun (_ : string) ->
             let msg =
-              In.Send
-                ( dest_queue_name
-                , {Message.payload= x; kind= Message.Request c.reply_queue_name}
+              Tracing.with_tracing
+                ~attributes:
+                  [
+                    ("messaging.operation.name", "send")
+                  ; ("messaging.system", "message-switch")
+                  ; ("messaging.destination.name", dest_queue_name)
+                  ]
+                ~span_kind:Producer ~parent:_span_parent
+                ~name:("send" ^ " " ^ dest_queue_name)
+                (fun _ ->
+                  In.Send
+                    ( dest_queue_name
+                    , {
+                        Message.payload= x
+                      ; kind= Message.Request c.reply_queue_name
+                      }
+                    )
                 )
             in
             do_rpc c.requests_conn msg >>|= fun (id : string) ->

ocaml/xapi-idl/lib/debug_info.ml

@@ -76,7 +76,7 @@ let to_log_string t = t.log
 
 (* Sets the logging context based on `dbg`.
    Also adds a new tracing span, linked to the parent span from `dbg`, if available. *)
-let with_dbg ?(with_thread = false) ~module_name ~name ~dbg f =
+let with_dbg ?(with_thread = false) ?(module_name = "") ~name ~dbg f =
   let di = of_string dbg in
   let f_with_trace () =
     let name =

ocaml/xapi-idl/lib/debug_info.mli

@@ -24,7 +24,7 @@ val to_log_string : t -> string
 
 val with_dbg :
      ?with_thread:bool
-  -> module_name:string
+  -> ?module_name:string
   -> name:string
   -> dbg:string
   -> (t -> 'a)

ocaml/xapi-idl/lib/xcp_client.ml

@@ -38,10 +38,35 @@ let switch_rpc ?timeout queue_name string_of_call response_of_string =
     get_ok
       (Message_switch_unix.Protocol_unix.Client.connect ~switch:!switch_path ())
   in
-  fun call ->
+  fun (call : Rpc.call) ->
+    let _span_parent =
+      call.params
+      |> List.find_map (function Rpc.Dict kv_list -> Some kv_list | _ -> None)
+      |> Fun.flip Option.bind
+           (List.find_map (function
+             | "debug_info", Rpc.String debug_info ->
+                 let di = debug_info |> Debug_info.of_string in
+                 di.tracing
+             | _ ->
+                 None
+             )
+             )
+    in
+    let rpc_service = "message_switch" in
+    Tracing.with_tracing
+      ~attributes:
+        [
+          ("rpc.system", "ocaml-rpc")
+        ; ("rpc.service", rpc_service)
+        ; ("server.address", queue_name)
+        ; ("rpc.method", call.name)
+        ]
+      ~parent:_span_parent
+      ~name:(rpc_service ^ "/" ^ call.name)
+    @@ fun _span_parent ->
     response_of_string
       (get_ok
-         (Message_switch_unix.Protocol_unix.Client.rpc ~t ?timeout
+         (Message_switch_unix.Protocol_unix.Client.rpc ?_span_parent ~t ?timeout
             ~queue:queue_name ~body:(string_of_call call) ()
          )
       )

ocaml/xenopsd/lib/xenops_server.ml

@@ -3682,7 +3682,9 @@ end
 module VM = struct
   module DB = VM_DB
 
-  let add _ dbg x = Debug.with_thread_associated dbg (fun () -> DB.add' x) ()
+  let add _ dbg x =
+    Debug_info.with_dbg ~with_thread:true ~name:__FUNCTION__ ~dbg @@ fun _ ->
+    DB.add' x
 
   let rename _ dbg id1 id2 when' =
     queue_operation dbg id1 (Atomic (VM_rename (id1, id2, when')))
@@ -3719,11 +3721,17 @@ module VM = struct
     in
     (vm_t, state)
 
-  let stat _ dbg id = Debug.with_thread_associated dbg (fun () -> stat' id) ()
+  let stat _ dbg id =
+    Debug_info.with_dbg ~with_thread:true ~name:__FUNCTION__ ~dbg @@ fun _ ->
+    stat' id
 
-  let exists _ _dbg id = match DB.read id with Some _ -> true | None -> false
+  let exists _ dbg id =
+    Debug_info.with_dbg ~name:__FUNCTION__ ~dbg @@ fun _ ->
+    match DB.read id with Some _ -> true | None -> false
 
-  let list _ dbg () = Debug.with_thread_associated dbg (fun () -> DB.list ()) ()
+  let list _ dbg () =
+    Debug_info.with_dbg ~with_thread:true ~name:__FUNCTION__ ~dbg @@ fun _ ->
+    DB.list ()
 
   let create _ dbg id =
     let no_sharept = false in