CP-50537: TGroup library to manage the priority and classify xapi execution threads. (#6076)

This is the follow up to
#6020

The initial phase is to classify the threads between Internal and
External.
- External is the default (for now),
- Internal are threads that process internal requests coming from smapi.

BVT + BST: 207007 (Dev Run)

Author: edwintorok

dune-project

@@ -37,6 +37,13 @@
   )
 )
 
+(package
+  (name tgroup)
+  (depends 
+    xapi-log
+    xapi-stdext-unix)
+)
+
 (package
   (name xml-light2)
 )
@@ -373,6 +380,7 @@
     tar
     tar-unix
     uri
+    tgroup
     (uuid (= :version))
     uutf
     uuidm
@@ -585,6 +593,7 @@ This package provides an Lwt compatible interface to the library.")
     (safe-resources(= :version))
     sha
     (stunnel (= :version))
+    tgroup
     uri
     (uuid (= :version))
     xapi-backtrace

http-lib.opam

@@ -22,6 +22,7 @@ depends: [
   "safe-resources" {= version}
   "sha"
   "stunnel" {= version}
+  "tgroup"
   "uri"
   "uuid" {= version}
   "xapi-backtrace"

ocaml/libs/http-lib/dune

@@ -44,6 +44,7 @@
     http_lib
     ipaddr
     polly
+    tgroup
     threads.posix
     tracing
     uri

ocaml/libs/http-lib/http.ml

@@ -132,6 +132,8 @@ module Hdr = struct
 
   let location = "location"
 
+  let originator = "originator"
+
   let traceparent = "traceparent"
 
   let hsts = "strict-transport-security"
@@ -688,6 +690,14 @@ module Request = struct
     let frame_header = if x.frame then make_frame_header headers else "" in
     frame_header ^ headers ^ body
 
+  let with_originator_of req f =
+    Option.iter
+      (fun req ->
+        let originator = List.assoc_opt Hdr.originator req.additional_headers in
+        f originator
+      )
+      req
+
   let traceparent_of req =
     let open Tracing in
     let ( let* ) = Option.bind in

ocaml/libs/http-lib/http.mli

@@ -129,6 +129,8 @@ module Request : sig
   val to_wire_string : t -> string
   (** [to_wire_string t] returns a string which could be sent to a server *)
 
+  val with_originator_of : t option -> (string option -> unit) -> unit
+
   val traceparent_of : t -> Tracing.Span.t option
 
   val with_tracing :

ocaml/libs/http-lib/http_svr.ml

@@ -561,6 +561,8 @@ let handle_connection ~header_read_timeout ~header_total_timeout
         ~max_length:max_header_length ss
     in
 
+    Http.Request.with_originator_of req Tgroup.of_req_originator ;
+
     (* 2. now we attempt to process the request *)
     let finished =
       Option.fold ~none:true

ocaml/libs/tgroup/dune

@@ -0,0 +1,4 @@
+(library
+ (name tgroup)
+ (public_name tgroup)
+ (libraries xapi-log xapi-stdext-unix))

ocaml/libs/tgroup/tgroup.ml

@@ -0,0 +1,184 @@
+(*
+ * 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.
+ *)
+
+module D = Debug.Make (struct let name = __MODULE__ end)
+
+open D
+
+let ( // ) = Filename.concat
+
+module Group = struct
+  module Internal = struct
+    type t
+
+    let name = "internal"
+  end
+
+  module External = struct
+    type t
+
+    let name = "external"
+  end
+
+  module Host = struct
+    type t
+
+    let name = "host"
+  end
+
+  module SM = struct
+    type t
+
+    let name = "SM"
+  end
+
+  type _ group =
+    | Internal_Host_SM : (Internal.t * Host.t * SM.t) group
+    | EXTERNAL : External.t group
+
+  type t = Group : 'a group -> t
+
+  let all = [Group Internal_Host_SM; Group EXTERNAL]
+
+  module Originator = struct
+    type t = Internal_Host_SM | EXTERNAL
+
+    let of_string = function
+      | s
+        when String.equal
+               (String.lowercase_ascii SM.name)
+               (String.lowercase_ascii s) ->
+          Internal_Host_SM
+      | s
+        when String.equal
+               (String.lowercase_ascii External.name)
+               (String.lowercase_ascii s) ->
+          EXTERNAL
+      | _ ->
+          EXTERNAL
+
+    let to_string = function
+      | Internal_Host_SM ->
+          SM.name
+      | EXTERNAL ->
+          External.name
+  end
+
+  module Creator = struct
+    type t = {
+        user: string option
+      ; endpoint: string option
+      ; originator: Originator.t
+    }
+
+    let make ?user ?endpoint originator = {originator; user; endpoint}
+
+    let to_string c =
+      Printf.sprintf "Creator -> user:%s endpoint:%s originator:%s"
+        (Option.value c.user ~default:"")
+        (Option.value c.endpoint ~default:"")
+        (Originator.to_string c.originator)
+  end
+
+  let of_originator = function
+    | Originator.Internal_Host_SM ->
+        Group Internal_Host_SM
+    | Originator.EXTERNAL ->
+        Group EXTERNAL
+
+  let get_originator = function
+    | Group Internal_Host_SM ->
+        Originator.Internal_Host_SM
+    | Group EXTERNAL ->
+        Originator.EXTERNAL
+
+  let of_creator creator = of_originator creator.Creator.originator
+
+  let to_cgroup : type a. a group -> string = function
+    | Internal_Host_SM ->
+        Internal.name // Host.name // SM.name
+    | EXTERNAL ->
+        External.name
+end
+
+module Cgroup = struct
+  type t = string
+
+  let cgroup_dir = Atomic.make None
+
+  let dir_of group : t option =
+    match group with
+    | Group.Group group ->
+        Option.map
+          (fun dir -> dir // Group.to_cgroup group)
+          (Atomic.get cgroup_dir)
+
+  let write_cur_tid_to_cgroup_file filename =
+    try
+      let perms = 0o640 in
+      let mode = [Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC] in
+      Xapi_stdext_unix.Unixext.with_file filename mode perms @@ fun fd ->
+      (* Writing 0 to the task file will automatically transform in writing
+         the current caller tid to the file.
+
+         Writing 0 to the processes file will automatically write the caller's
+         pid to file. *)
+      let buf = "0\n" in
+      let len = String.length buf in
+      if Unix.write fd (Bytes.unsafe_of_string buf) 0 len <> len then
+        warn "writing current tid to %s failed" filename
+    with exn ->
+      warn "writing current tid to %s failed with exception: %s" filename
+        (Printexc.to_string exn)
+
+  let attach_task group =
+    Option.iter
+      (fun dir ->
+        let tasks_file = dir // "tasks" in
+        write_cur_tid_to_cgroup_file tasks_file
+      )
+      (dir_of group)
+
+  let set_cur_cgroup ~originator =
+    match originator with
+    | Group.Originator.Internal_Host_SM ->
+        attach_task (Group Internal_Host_SM)
+    | Group.Originator.EXTERNAL ->
+        attach_task (Group EXTERNAL)
+
+  let set_cgroup creator =
+    set_cur_cgroup ~originator:creator.Group.Creator.originator
+
+  let init dir =
+    let () = Atomic.set cgroup_dir (Some dir) in
+    Group.all
+    |> List.filter_map dir_of
+    |> List.iter (fun dir -> Xapi_stdext_unix.Unixext.mkdir_rec dir 0o755) ;
+    set_cur_cgroup ~originator:Group.Originator.EXTERNAL
+end
+
+let of_originator originator =
+  originator |> Group.Creator.make |> Cgroup.set_cgroup
+
+let of_req_originator originator =
+  Option.iter
+    (fun _ ->
+      try
+        originator
+        |> Option.value ~default:Group.Originator.(to_string EXTERNAL)
+        |> Group.Originator.of_string
+        |> of_originator
+      with _ -> ()
+    )
+    (Atomic.get Cgroup.cgroup_dir)

ocaml/libs/tgroup/tgroup.mli

@@ -0,0 +1,78 @@
+(*
+ * 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.
+ *)
+
+(** [Group] module helps with the classification of different xapi execution
+    threads.*)
+module Group : sig
+  (** Abstract type that represents a group of execution threads in xapi. Each
+  group corresponds to a Creator, and has a designated level of priority.*)
+  type t
+
+  (** Generic representation of different xapi threads originators. *)
+  module Originator : sig
+    (** Type that represents different originators of xapi threads. *)
+    type t = Internal_Host_SM | EXTERNAL
+
+    val of_string : string -> t
+    (** [of_string s] creates an originator from a string [s].
+    
+    e.g create an originator based on a http header. *)
+
+    val to_string : t -> string
+    (** [to_string o] converts an originator [o] to its string representation.*)
+  end
+
+  (** Generic representation of different xapi threads creators. *)
+  module Creator : sig
+    (** Abstract type that represents different creators of xapi threads.*)
+    type t
+
+    val make : ?user:string -> ?endpoint:string -> Originator.t -> t
+    (** [make o] creates a creator type based on a given originator [o].*)
+
+    val to_string : t -> string
+    (** [to_string c] converts a creator [c] to its string representation.*)
+  end
+
+  val get_originator : t -> Originator.t
+  (** [get_originator group] returns the originator that maps to group [group].*)
+
+  val of_creator : Creator.t -> t
+  (** [of_creator c] returns the corresponding group based on the creator [c].*)
+end
+
+(** [Cgroup] module encapsulates different function for managing the cgroups
+corresponding with [Groups].*)
+module Cgroup : sig
+  (** Represents one of the children of the cgroup directory.*)
+  type t = string
+
+  val dir_of : Group.t -> t option
+  (** [dir_of group] returns the full path of the cgroup directory corresponding
+      to the group [group] as [Some dir].
+      
+      Returns [None] if [init dir] has not been called. *)
+
+  val init : string -> unit
+  (** [init dir] initializes the hierachy of cgroups associated to all [Group.t]
+      types under the directory [dir].*)
+
+  val set_cgroup : Group.Creator.t -> unit
+  (** [set_cgroup c] sets the current xapi thread in a cgroup based on the
+      creator [c].*)
+end
+
+val of_req_originator : string option -> unit
+(** [of_req_originator o] same as [of_originator] but it classifies based on the
+http request header.*)

ocaml/xapi/dune

@@ -151,6 +151,7 @@
     tapctl
     tar
     tar-unix
+    tgroup
     threads.posix
     tracing
     unixpwd
@@ -237,6 +238,7 @@
     rpclib.json
     rpclib.xml
     stunnel
+    tgroup
     threads.posix
     tracing
     xapi-backtrace

ocaml/xapi/sm_exec.ml

@@ -38,6 +38,13 @@ let with_dbg ~name ~dbg f =
 (*********************************************************************************************)
 (* Random utility functions *)
 
+let env_vars =
+  Array.concat
+    [
+      Forkhelpers.default_path_env_pair
+    ; Env_record.to_string_array [Env_record.pair ("ORIGINATOR", "SM")]
+    ]
+
 type call = {
     (* All calls are performed by a specific Host with a special Session and device_config *)
     host_ref: API.ref_host
@@ -355,9 +362,9 @@ let exec_xmlrpc ~dbg ?context:_ ?(needs_session = true) (driver : string)
               let env, exe, args =
                 match Xapi_observer_components.is_smapi_enabled () with
                 | false ->
-                    (None, exe, args)
+                    (Some env_vars, exe, args)
                 | true ->
-                    Xapi_observer_components.env_exe_args_of
+                    Xapi_observer_components.env_exe_args_of ~env_vars
                       ~component:Xapi_observer_components.SMApi ~exe ~args
               in
               Forkhelpers.execute_command_get_output ?tracing:di.tracing ?env