runtime events collector, + test

dune-project

@@ -91,4 +91,19 @@
  (tags
   (trace tracing fuchsia)))
 
+(package
+ (name trace-runtime-events)
+ (synopsis
+  "A simple collector relying on runtime-events for OCaml 5. Some assembly required.")
+ (depends
+  (ocaml
+   (>= 5.0))
+  (trace
+   (= :version))
+  base-bigarray
+  base-unix
+  dune)
+ (tags
+  (trace tracing trace runtime-events)))
+
 ; See the complete stanza docs at https://dune.readthedocs.io/en/stable/dune-files.html#dune-project

src/runtime-events/dune

@@ -0,0 +1,8 @@
+(library
+ (name trace_runtime_events)
+ (public_name trace-runtime-events)
+ (synopsis "Simple collector that emits events via Runtime_events")
+ (libraries
+  trace.core
+  trace.util
+  (re_export runtime_events)))

src/runtime-events/trace_runtime_events.ml

@@ -0,0 +1,151 @@
+(** Simple backend that emits trace events via Runtime_events.
+
+    This backend allows trace spans, messages, and metrics to be collected by
+    external tools using the OCaml Runtime_events system. *)
+
+open Trace_core
+
+(* Register custom event types for strings *)
+module String_type = struct
+  let max_len = 1024
+
+  let encode buf s =
+    let len = min (String.length s) (max_len - 1) in
+    Bytes.blit_string s 0 buf 0 len;
+    len
+
+  let decode buf len = Bytes.sub_string buf 0 len
+  let ty = Runtime_events.Type.register ~encode ~decode
+end
+
+module String_int = struct
+  let max_len = 1024
+
+  let encode buf (s, i) =
+    let len = min (String.length s) (max_len - 9) in
+    Bytes.set_int64_le buf 0 (Int64.of_int i);
+    Bytes.blit_string s 0 buf 8 len;
+    len + 8
+
+  let decode buf len =
+    let i = Bytes.get_int64_le buf 0 in
+    Bytes.sub_string buf 8 (len - 8), Int64.to_int i
+
+  let ty = Runtime_events.Type.register ~encode ~decode
+end
+
+module String_float = struct
+  let max_len = 1024
+
+  let encode buf (s, f) =
+    let len = min (String.length s) (max_len - 9) in
+    Bytes.set_int64_le buf 0 (Int64.bits_of_float f);
+    Bytes.blit_string s 0 buf 8 len;
+    len + 8
+
+  let decode buf len =
+    let i = Bytes.get_int64_le buf 0 in
+    Bytes.sub_string buf 8 (len - 8), Int64.float_of_bits i
+
+  let ty = Runtime_events.Type.register ~encode ~decode
+end
+
+module Events = struct
+  (* Define event tags *)
+  type Runtime_events.User.tag +=
+    | Tag_span_enter
+    | Tag_span_exit
+    | Tag_message
+    | Tag_metric_int
+    | Tag_metric_float
+
+  (* Register user events *)
+  let span_enter_event =
+    Runtime_events.User.register "trace.span.enter" Tag_span_enter
+      String_type.ty
+
+  let span_exit_event =
+    Runtime_events.User.register "trace.span.exit" Tag_span_exit String_type.ty
+
+  let message_event =
+    Runtime_events.User.register "trace.message" Tag_message String_type.ty
+
+  let metric_int_event =
+    Runtime_events.User.register "trace.metric.int" Tag_metric_int String_int.ty
+
+  let metric_float_event =
+    Runtime_events.User.register "trace.metric.float" Tag_metric_float
+      String_float.ty
+end
+
+(* Span representation *)
+type span_info = { name: string }
+type Trace_core.span += Span_runtime_events of span_info
+
+(* Collector state *)
+type st = {
+  active: bool Trace_core.Internal_.Atomic_.t;
+  start_events: bool;
+}
+
+let create ?(start_events = true) () : st =
+  { active = Trace_core.Internal_.Atomic_.make true; start_events }
+
+(* Collector callbacks *)
+let init (self : st) = if self.start_events then Runtime_events.start ()
+
+let shutdown (self : st) =
+  Trace_core.Internal_.Atomic_.set self.active false;
+  Runtime_events.pause ()
+
+let enabled _ _ = true
+
+let enter_span (_self : st) ~__FUNCTION__:_ ~__FILE__:_ ~__LINE__:_ ~level:_
+    ~params:_ ~data:_ ~parent:_ name : span =
+  Runtime_events.User.write Events.span_enter_event name;
+  Span_runtime_events { name }
+
+let exit_span (_self : st) sp =
+  match sp with
+  | Span_runtime_events info ->
+    Runtime_events.User.write Events.span_exit_event info.name
+  | _ -> ()
+
+let add_data_to_span _st _sp _data =
+  (* Runtime_events doesn't support adding data to spans after creation,
+     so we just ignore this *)
+  ()
+
+let message (_self : st) ~level:_ ~params:_ ~data:_ ~span:_ msg : unit =
+  Runtime_events.User.write Events.message_event msg
+
+let metric (_self : st) ~level:_ ~params:_ ~data:_ name m : unit =
+  match m with
+  | Core_ext.Metric_int n ->
+    Runtime_events.User.write Events.metric_int_event (name, n)
+  | Core_ext.Metric_float f ->
+    Runtime_events.User.write Events.metric_float_event (name, f)
+  | _ -> ()
+
+let extension _self ~level:_ _ev =
+  (* Extension events like set_thread_name, set_process_name could be
+     emitted as custom events if needed *)
+  ()
+
+(* Create collector *)
+let callbacks : st Collector.Callbacks.t =
+  Collector.Callbacks.make ~init ~shutdown ~enabled ~enter_span ~exit_span
+    ~add_data_to_span ~message ~metric ~extension ()
+
+let collector ?(start_events = true) () : Collector.t =
+  let st = create ~start_events () in
+  Collector.C_some (st, callbacks)
+
+(* Setup function *)
+let setup ?(start_events = true) () =
+  Trace_core.setup_collector (collector ~start_events ())
+
+(* Convenience wrapper *)
+let with_setup ?start_events f =
+  setup ?start_events ();
+  Fun.protect ~finally:Trace_core.shutdown f

src/runtime-events/trace_runtime_events.mli

@@ -0,0 +1,55 @@
+(** Simple collector that emits trace events via Runtime_events.
+
+    @since NEXT_RELEASE *)
+
+(** {2 Event types for decoding} *)
+
+module String_type : sig
+  val ty : string Runtime_events.Type.t
+end
+
+module String_int : sig
+  val ty : (string * int) Runtime_events.Type.t
+end
+
+module String_float : sig
+  val ty : (string * float) Runtime_events.Type.t
+end
+
+(** Custom events *)
+module Events : sig
+  type Runtime_events.User.tag +=
+    | Tag_span_enter
+    | Tag_span_exit
+    | Tag_message
+    | Tag_metric_int
+    | Tag_metric_float
+
+  val span_enter_event : string Runtime_events.User.t
+  val span_exit_event : string Runtime_events.User.t
+  val message_event : string Runtime_events.User.t
+  val metric_int_event : (string * int) Runtime_events.User.t
+  val metric_float_event : (string * float) Runtime_events.User.t
+end
+
+(** {2 Collector} *)
+
+val collector : ?start_events:bool -> unit -> Trace_core.Collector.t
+(** [collector ~start_events ()] creates a new collector that emits events via
+    Runtime_events.
+
+    @param start_events
+      if [true] (default), automatically call [Runtime_events.start()] when the
+      collector is initialized. *)
+
+val setup : ?start_events:bool -> unit -> unit
+(** [setup ~start_events ()] sets up the Runtime_events collector as the global
+    collector.
+
+    See {!collector} *)
+
+val with_setup : ?start_events:bool -> (unit -> 'a) -> 'a
+(** [with_setup ~start_events f] runs [f ()] with the Runtime_events collector
+    enabled, and shuts it down when done.
+
+    See {!collector} *)

test/dune

@@ -17,3 +17,10 @@
  (modules t_debug)
  (package trace-tef)
  (libraries trace trace.debug trace-tef))
+
+(test
+ (name t_runtime_events)
+ (modules t_runtime_events)
+ (preprocess
+  (pps ppx_trace))
+ (libraries trace trace-runtime-events runtime_events))

test/t_runtime_events.ml

@@ -0,0 +1,164 @@
+(** Test program for the Runtime_events backend.
+
+    This demonstrates and tests the Runtime_events backend by: 1. Emitting trace
+    events 2. Subscribing to the ringbuffer 3. Collecting all events 4.
+    Verifying expected events were emitted *)
+
+let ( let@ ) = ( @@ )
+
+(* Event types we'll collect *)
+type collected_event =
+  | Span_enter of string
+  | Span_exit of string
+  | Message of string
+  | Metric_int of string * int
+  | Metric_float of string * float
+
+(* Simple recursive function to generate some trace activity *)
+let rec fib x =
+  let%trace () = "fib" in
+  if x <= 2 then
+    1
+  else
+    fib (x - 1) + fib (x - 2)
+
+(* Function with explicit span *)
+let do_work () =
+  Trace_core.with_span ~__FILE__ ~__LINE__ "do_work" @@ fun _sp ->
+  Trace_core.message "Starting work";
+  Trace_core.counter_int "work_units" 100;
+
+  let result = fib 10 in
+
+  Trace_core.messagef (fun k -> k "Computed fib(10) = %d" result);
+  Trace_core.counter_int "work_units" 200;
+  result
+
+(* Subscribe to runtime events and collect them *)
+let collect_events () =
+  let events = ref [] in
+
+  (* Create a cursor to read from our own process *)
+  let cursor = Runtime_events.create_cursor None in
+
+  (* Set up callbacks *)
+  let callbacks =
+    Runtime_events.Callbacks.create ()
+    (* Register callbacks for our custom events using type values *)
+    |> Runtime_events.Callbacks.add_user_event
+         Trace_runtime_events.String_type.ty (fun _domain_id _ts tag name ->
+           match Runtime_events.User.tag tag with
+           | Trace_runtime_events.Events.Tag_span_enter ->
+             events := Span_enter name :: !events
+           | Trace_runtime_events.Events.Tag_span_exit ->
+             events := Span_exit name :: !events
+           | Trace_runtime_events.Events.Tag_message ->
+             events := Message name :: !events
+           | _ -> ())
+    |> Runtime_events.Callbacks.add_user_event
+         Trace_runtime_events.String_int.ty
+         (fun _domain_id _ts tag (name, value) ->
+           match Runtime_events.User.tag tag with
+           | Trace_runtime_events.Events.Tag_metric_int ->
+             events := Metric_int (name, value) :: !events
+           | _ -> ())
+    |> Runtime_events.Callbacks.add_user_event
+         Trace_runtime_events.String_float.ty
+         (fun _domain_id _ts tag (name, value) ->
+           match Runtime_events.User.tag tag with
+           | Trace_runtime_events.Events.Tag_metric_float ->
+             events := Metric_float (name, value) :: !events
+           | _ -> ())
+  in
+
+  (* Read all events from the ringbuffer *)
+  let _lost_events = Runtime_events.read_poll cursor callbacks None in
+
+  List.rev !events
+
+let () =
+  (* Initialize the Runtime_events backend with start_events=false
+     so we can manually control when to start *)
+  Trace_runtime_events.setup ~start_events:false ();
+
+  (* Start runtime events *)
+  Runtime_events.start ();
+
+  (* Set process and thread names *)
+  Trace_core.set_process_name "test";
+  Trace_core.set_thread_name "main";
+
+  (* Do some traced work *)
+  let result = do_work () in
+  Printf.eprintf "result: %d\n" result;
+
+  (* Collect events from the ringbuffer *)
+  let events = collect_events () in
+
+  Printf.eprintf "\ncollected %d events:\n" (List.length events);
+  List.iter
+    (fun ev ->
+      match ev with
+      | Span_enter name -> Printf.eprintf "  - span enter: %s\n" name
+      | Span_exit name -> Printf.eprintf "  - span exit: %s\n" name
+      | Message msg -> Printf.eprintf "  - message: %s\n" msg
+      | Metric_int (name, value) ->
+        Printf.eprintf "  - metric int: %s = %d\n" name value
+      | Metric_float (name, value) ->
+        Printf.eprintf "  - metric float: %s = %f\n" name value)
+    events;
+
+  (* Verify expected events *)
+  let has_do_work_enter =
+    List.exists
+      (function
+        | Span_enter "do_work" -> true
+        | _ -> false)
+      events
+  in
+  let has_do_work_exit =
+    List.exists
+      (function
+        | Span_exit "do_work" -> true
+        | _ -> false)
+      events
+  in
+  let has_fib_spans =
+    List.filter
+      (function
+        | Span_enter "fib" | Span_exit "fib" -> true
+        | _ -> false)
+      events
+  in
+  let has_starting_work =
+    List.exists
+      (function
+        | Message "Starting work" -> true
+        | _ -> false)
+      events
+  in
+  let has_metrics =
+    List.filter
+      (function
+        | Metric_int ("work_units", _) -> true
+        | _ -> false)
+      events
+  in
+
+  Printf.eprintf "\nVerification:\n";
+  Printf.eprintf "  - do_work span enter: %b\n" has_do_work_enter;
+  Printf.eprintf "  - do_work span exit: %b\n" has_do_work_exit;
+  Printf.eprintf "  - fib spans (enter+exit): %d\n" (List.length has_fib_spans);
+  Printf.eprintf "  - 'Starting work' message: %b\n" has_starting_work;
+  Printf.eprintf "  - work_units metrics: %d\n" (List.length has_metrics);
+
+  (* Check assertions *)
+  assert has_do_work_enter;
+  assert has_do_work_exit;
+  assert (List.length has_fib_spans = 218);
+  assert has_starting_work;
+  assert (List.length has_metrics = 2);
+
+  Printf.eprintf "\nall good :-)\n";
+
+  Trace_core.shutdown ()

trace-runtime-events.opam

@@ -0,0 +1,36 @@
+# This file is generated by dune, edit dune-project instead
+opam-version: "2.0"
+version: "0.11"
+synopsis:
+  "A simple collector relying on runtime-events for OCaml 5. Some assembly required."
+maintainer: ["Simon Cruanes"]
+authors: ["Simon Cruanes"]
+license: "MIT"
+tags: ["trace" "tracing" "trace" "runtime-events"]
+homepage: "https://github.com/ocaml-tracing/ocaml-trace"
+bug-reports: "https://github.com/ocaml-tracing/ocaml-trace/issues"
+depends: [
+  "ocaml" {>= "5.0"}
+  "trace" {= version}
+  "base-bigarray"
+  "base-unix"
+  "dune" {>= "2.9"}
+  "odoc" {with-doc}
+]
+build: [
+  ["dune" "subst"] {dev}
+  [
+    "dune"
+    "build"
+    "-p"
+    name
+    "-j"
+    jobs
+    "--promote-install-files=false"
+    "@install"
+    "@runtest" {with-test}
+    "@doc" {with-doc}
+  ]
+  ["dune" "install" "-p" name "--create-install-files" name]
+]
+dev-repo: "git+https://github.com/ocaml-tracing/ocaml-trace.git"