gazagnaire.org
Logging, runtime tracing, and the obs performance profiler for OCaml
obs: split the observe library from the obs CLI
One opam package shipped both, so a program that wanted Cmdliner logging terms declared a dependency on sqlite, hdr, uuidm, cstruct, eio_main and monitor. Those belong to the profiler. obs now carries the CLI, its collector and its trace adapters; observe carries the library. observe also drops nox-tty, since Tty.Display now installs the log-suspending reporter wrapper itself.
Dune ties a public_name to its package, so every observe.* sublibrary stays in observe, and the CLI's four libraries take public names under obs: obs.model, obs.fxt, obs.memtrace, obs.offline. They had to become public because dune forbids a public library depending on a private one, and being public brought their modules under merlint's untested-module rule. So they are tested now: 18 cases over the Fuchsia trace codec and 24 over the collector, its adapters and the CTF reader, each with a hostile case the public API must reject.
observe.opam still declares nox-protobuf, nox-varint, bytesrw and vsock through observe.producer, and nox-json and nox-catapult through observe.capture. Neither is the logging library's business; both are addressable by moving those sublibraries into packages of their own.
author
Thomas Gazagnaire
date
(Jul 9, 2026, 11:29 PM UTC)
commit
8c6e8a34
8c6e8a34148435ef497b856c6cd6e9609e9bd7d2
parent
cb26f7f8
cb26f7f8da60309a693b98f8d7b7ac2e531943e6
+1183
-24
+2
-2
bin/obs/dune
+2
-2
bin/obs/dune
+1
dune
+1
dune
+28
-6
dune-project
+28
-6
dune-project
···
20
20
21
21
(package
22
22
(name observe)
23
-
(synopsis "Logging, runtime tracing, and the obs performance profiler for OCaml")
23
+
(synopsis "Cmdliner logging terms and one-call setup for OCaml programs")
24
24
(tags (org:blacksun logging cli system))
25
25
(description
26
-
"observe is an OCaml observability toolkit. The Observe library provides Cmdliner terms for structured logging -- verbosity flags (-q, -v, -vv, -vvv), RUST_LOG-style filtering (--log=level,src:level), JSON logs (--json), and *.tracing log sources (--trace FILE) -- and honors MEMTRACE for allocation capture. The obs CLI captures a program's runtime_events scheduler/GC trace, memtrace allocation traces, and ocaml-probe spans, then reports a wall-clock budget and verdict: it writes Fuchsia (.fxt) and Catapult/Perfetto (.pb) traces and a Perfetto-schema SQLite database (queried with obs query), drills allocation hotspots (obs hotspots), and profiles CPU-bound code with the native sampler (obs run --sample). A producer/collector protocol streams traces from subprocesses and VM guests into a central obs collector.")
26
+
"The Observe library provides Cmdliner terms for structured logging: verbosity flags (-q, -v, -vv, -vvv), RUST_LOG-style filtering (--log=level,src:level), and JSON logs (--json). A single call to Observe.setup wires these terms into a program and also makes it instrumentable: it honours the MEMTRACE environment variable to capture an allocation trace, adopts a W3C traceparent so its spans join a distributed trace, and streams runtime_events to an obs collector when OBS_COLLECTOR is set.")
27
27
(depends
28
28
(ocaml (>= 5.2))
29
29
(logs (>= 0.7))
···
37
37
nox-catapult
38
38
nox-protobuf
39
39
nox-varint
40
-
sqlite
41
40
json-logs
42
41
probe
43
-
uuidm
42
+
(eio (>= 1.0))
43
+
vsock
44
+
(mdx :with-test)
45
+
(alcotest :with-test)))
46
+
47
+
(package
48
+
(name obs)
49
+
(synopsis "Performance profiler that captures and reports OCaml runtime traces")
50
+
(tags (org:blacksun cli system))
51
+
(description
52
+
"obs captures a program's runtime_events scheduler and GC trace, its memtrace allocation trace, and ocaml-probe spans, then reports a wall-clock budget and a verdict. It writes Fuchsia (.fxt) and Catapult/Perfetto (.pb) traces and a Perfetto-schema SQLite database: obs query runs SQL over a captured trace, obs hotspots drills allocation sites, and obs run --sample profiles CPU-bound code with the native sampler.")
53
+
(depends
54
+
(ocaml (>= 5.2))
55
+
(observe (= :version))
56
+
(logs (>= 0.7))
57
+
(fmt (>= 0.9))
58
+
base-unix
59
+
(cmdliner (>= 1.2))
60
+
(mtime (>= 2.0))
61
+
bytesrw
62
+
nox-json
63
+
nox-catapult
64
+
probe
44
65
hdr
66
+
uuidm
67
+
cstruct
68
+
sqlite
45
69
(eio (>= 1.0))
46
70
(eio_main (>= 1.0))
47
71
vsock
48
-
cstruct
49
-
(mdx :with-test)
50
72
nox-tty
51
73
(monitor :with-test)
52
74
(alcotest :with-test)))
+2
-2
lib/dune
+2
-2
lib/dune
+1
lib/obs/dune
+1
lib/obs/dune
+1
lib/obs/fxt/dune
+1
lib/obs/fxt/dune
+1
lib/obs/memtrace/dune
+1
lib/obs/memtrace/dune
+1
lib/obs/offline/dune
+1
lib/obs/offline/dune
+2
-2
lib/observe.ml
+2
-2
lib/observe.ml
···
251
251
in
252
252
Fmt_tty.setup_std_outputs ();
253
253
Logs.set_level global_level;
254
-
Logs.set_reporter (Tty.Display.logs_reporter (Logs_fmt.reporter ()));
254
+
Logs.set_reporter (Logs_fmt.reporter ());
255
255
apply_source_overrides source_overrides
256
256
257
257
type log_flags = { quiet : bool; json : bool }
···
328
328
{ Logs.report }
329
329
| None -> main_reporter
330
330
in
331
-
Logs.set_reporter (Tty.Display.logs_reporter reporter);
331
+
Logs.set_reporter reporter;
332
332
(* Set global level: --log override > -q/-v flags *)
333
333
let level =
334
334
match global_override with
+60
obs.opam
+60
obs.opam
···
1
+
# This file is generated by dune, edit dune-project instead
2
+
opam-version: "2.0"
3
+
synopsis:
4
+
"Performance profiler that captures and reports OCaml runtime traces"
5
+
description:
6
+
"obs captures a program's runtime_events scheduler and GC trace, its memtrace allocation trace, and ocaml-probe spans, then reports a wall-clock budget and a verdict. It writes Fuchsia (.fxt) and Catapult/Perfetto (.pb) traces and a Perfetto-schema SQLite database: obs query runs SQL over a captured trace, obs hotspots drills allocation sites, and obs run --sample profiles CPU-bound code with the native sampler."
7
+
maintainer: ["Thomas Gazagnaire <thomas@gazagnaire.org>"]
8
+
authors: ["Thomas Gazagnaire <thomas@gazagnaire.org>"]
9
+
license: "ISC AND MIT"
10
+
tags: ["org:blacksun" "cli" "system"]
11
+
homepage: "https://tangled.org/gazagnaire.org/ocaml-observe"
12
+
bug-reports: "https://tangled.org/gazagnaire.org/ocaml-observe/issues"
13
+
depends: [
14
+
"dune" {>= "3.21"}
15
+
"ocaml" {>= "5.2"}
16
+
"observe" {= version}
17
+
"logs" {>= "0.7"}
18
+
"fmt" {>= "0.9"}
19
+
"base-unix"
20
+
"cmdliner" {>= "1.2"}
21
+
"mtime" {>= "2.0"}
22
+
"bytesrw"
23
+
"nox-json"
24
+
"nox-catapult"
25
+
"probe"
26
+
"hdr"
27
+
"uuidm"
28
+
"cstruct"
29
+
"sqlite"
30
+
"eio" {>= "1.0"}
31
+
"eio_main" {>= "1.0"}
32
+
"vsock"
33
+
"nox-tty"
34
+
"monitor" {with-test}
35
+
"alcotest" {with-test}
36
+
"odoc" {with-doc}
37
+
]
38
+
build: [
39
+
["dune" "subst"] {dev}
40
+
[
41
+
"dune"
42
+
"build"
43
+
"-p"
44
+
name
45
+
"-j"
46
+
jobs
47
+
"@install"
48
+
"@runtest" {with-test}
49
+
"@doc" {with-doc}
50
+
]
51
+
]
52
+
dev-repo: "git+https://tangled.org/gazagnaire.org/ocaml-observe"
53
+
x-maintenance-intent: ["(latest)"]
54
+
pin-depends: [
55
+
["bytesrw.0.3.0" "git+https://github.com/dbuenzli/bytesrw.git#d62c9007aa28f8da872a9662a6e95f9979bca38b"]
56
+
["eio.dev" "git+https://github.com/ocaml-multicore/eio.git#e7e6cf38a643ff0665c96d9ef6e68a998663a246"]
57
+
["eio_main.dev" "git+https://github.com/ocaml-multicore/eio.git#e7e6cf38a643ff0665c96d9ef6e68a998663a246"]
58
+
["eio_posix.dev" "git+https://github.com/ocaml-multicore/eio.git#e7e6cf38a643ff0665c96d9ef6e68a998663a246"]
59
+
["cstruct.6.2.0" "git+https://github.com/samoht/ocaml-cstruct.git#d1e1f46bd0711d301a613411ef325160c0d8b21c"]
60
+
]
+7
obs.opam.template
+7
obs.opam.template
···
1
+
pin-depends: [
2
+
["bytesrw.0.3.0" "git+https://github.com/dbuenzli/bytesrw.git#d62c9007aa28f8da872a9662a6e95f9979bca38b"]
3
+
["eio.dev" "git+https://github.com/ocaml-multicore/eio.git#e7e6cf38a643ff0665c96d9ef6e68a998663a246"]
4
+
["eio_main.dev" "git+https://github.com/ocaml-multicore/eio.git#e7e6cf38a643ff0665c96d9ef6e68a998663a246"]
5
+
["eio_posix.dev" "git+https://github.com/ocaml-multicore/eio.git#e7e6cf38a643ff0665c96d9ef6e68a998663a246"]
6
+
["cstruct.6.2.0" "git+https://github.com/samoht/ocaml-cstruct.git#d1e1f46bd0711d301a613411ef325160c0d8b21c"]
7
+
]
+2
-10
observe.opam
+2
-10
observe.opam
···
1
1
# This file is generated by dune, edit dune-project instead
2
2
opam-version: "2.0"
3
-
synopsis:
4
-
"Logging, runtime tracing, and the obs performance profiler for OCaml"
3
+
synopsis: "Cmdliner logging terms and one-call setup for OCaml programs"
5
4
description:
6
-
"observe is an OCaml observability toolkit. The Observe library provides Cmdliner terms for structured logging -- verbosity flags (-q, -v, -vv, -vvv), RUST_LOG-style filtering (--log=level,src:level), JSON logs (--json), and *.tracing log sources (--trace FILE) -- and honors MEMTRACE for allocation capture. The obs CLI captures a program's runtime_events scheduler/GC trace, memtrace allocation traces, and ocaml-probe spans, then reports a wall-clock budget and verdict: it writes Fuchsia (.fxt) and Catapult/Perfetto (.pb) traces and a Perfetto-schema SQLite database (queried with obs query), drills allocation hotspots (obs hotspots), and profiles CPU-bound code with the native sampler (obs run --sample). A producer/collector protocol streams traces from subprocesses and VM guests into a central obs collector."
5
+
"The Observe library provides Cmdliner terms for structured logging: verbosity flags (-q, -v, -vv, -vvv), RUST_LOG-style filtering (--log=level,src:level), and JSON logs (--json). A single call to Observe.setup wires these terms into a program and also makes it instrumentable: it honours the MEMTRACE environment variable to capture an allocation trace, adopts a W3C traceparent so its spans join a distributed trace, and streams runtime_events to an obs collector when OBS_COLLECTOR is set."
7
6
maintainer: ["Thomas Gazagnaire <thomas@gazagnaire.org>"]
8
7
authors: ["Thomas Gazagnaire <thomas@gazagnaire.org>"]
9
8
license: "ISC AND MIT"
···
24
23
"nox-catapult"
25
24
"nox-protobuf"
26
25
"nox-varint"
27
-
"sqlite"
28
26
"json-logs"
29
27
"probe"
30
-
"uuidm"
31
-
"hdr"
32
28
"eio" {>= "1.0"}
33
-
"eio_main" {>= "1.0"}
34
29
"vsock"
35
-
"cstruct"
36
30
"mdx" {with-test}
37
-
"nox-tty"
38
-
"monitor" {with-test}
39
31
"alcotest" {with-test}
40
32
"odoc" {with-doc}
41
33
]
+1
test/capture/dune
+1
test/capture/dune
+1
test/cram/dune
+1
test/cram/dune
+1
test/dune
+1
test/dune
+1
-1
test/memtrace/dune
+1
-1
test/memtrace/dune
+4
test/obs/fxt/dune
+4
test/obs/fxt/dune
+1
test/obs/fxt/test.ml
+1
test/obs/fxt/test.ml
···
1
+
let () = Alcotest.run "obs-fxt" [ Test_read.suite; Test_write.suite ]
+254
test/obs/fxt/test_read.ml
+254
test/obs/fxt/test_read.ml
···
1
+
(* Tests for {!Observer_fxt.Read}, the Fuchsia trace format reader.
2
+
3
+
Format reference: https://fuchsia.dev/fuchsia-src/reference/tracing/trace-format
4
+
The reader is deliberately tolerant of a trace being written concurrently, so
5
+
a truncated tail is dropped rather than fabricated; these tests pin the exact
6
+
records recovered from valid input and the exact tolerant behaviour on
7
+
malformed input. *)
8
+
9
+
module Read = Observer_fxt.Read
10
+
module Write = Observer_fxt.Write
11
+
12
+
(* --- helpers --- *)
13
+
14
+
let to_bytes f =
15
+
let buf = Eio.Buf_write.create 0x1000 in
16
+
let w = Write.of_writer buf in
17
+
f w;
18
+
Eio.Buf_write.serialize_to_string buf
19
+
20
+
(* Parse a byte string into its record list. The records are collected inside
21
+
the parser callback, so a leftover-bytes error from a truncated tail (which
22
+
[parse_string] reports after the value is built) does not discard the records
23
+
already recovered -- exactly what the tolerant reader is meant to yield. *)
24
+
let parse data =
25
+
let got = ref [] in
26
+
(try
27
+
ignore
28
+
(Eio.Buf_read.parse_string
29
+
(fun r -> got := List.of_seq (Read.records r))
30
+
data)
31
+
with Failure _ -> ());
32
+
!got
33
+
34
+
let le64 buf (x : int64) =
35
+
let b = Bytes.create 8 in
36
+
Bytes.set_int64_le b 0 x;
37
+
Buffer.add_bytes buf b
38
+
39
+
let events records =
40
+
List.filter_map (function Read.Event e -> Some e | _ -> None) records
41
+
42
+
let thread : Write.thread = { pid = 7L; tid = 3L }
43
+
44
+
(* --- an empty stream and a stream with only the magic record --- *)
45
+
46
+
let test_empty () =
47
+
Alcotest.(check int) "no records in empty input" 0 (List.length (parse ""))
48
+
49
+
let test_magic_only () =
50
+
(* [of_writer] emits exactly the Fuchsia magic-number metadata record. *)
51
+
match parse (to_bytes (fun _ -> ())) with
52
+
| [ Read.Metadata ] -> ()
53
+
| rs -> Alcotest.failf "expected [Metadata], got %d records" (List.length rs)
54
+
55
+
(* --- round-trip: an instant event with every arg type --- *)
56
+
57
+
(* Writer arg types map to reader values exactly: [`Int64]->Int64,
58
+
[`String]->String, [`Pointer]->Pointer, [`Koid]->Koid. A [`Unit] arg has
59
+
wire type 0, which the reader has no case for, so it reads back as
60
+
[`Unknown 0]. Arg order is preserved. *)
61
+
let test_instant_all_args () =
62
+
let data =
63
+
to_bytes (fun w ->
64
+
Write.instant_event w ~name:"paint" ~thread ~category:"gfx" ~ts:1234L
65
+
~args:
66
+
[
67
+
("w", `Int64 42L);
68
+
("u", `Unit);
69
+
("s", `String "hi");
70
+
("p", `Pointer 16L);
71
+
("k", `Koid 99L);
72
+
])
73
+
in
74
+
match events (parse data) with
75
+
| [ ev ] -> (
76
+
Alcotest.(check bool) "type is instant" true (ev.ty = Read.Event.Instant);
77
+
Alcotest.(check int64) "timestamp" 1234L ev.timestamp;
78
+
Alcotest.(check int64) "thread pid" 7L ev.thread.pid;
79
+
Alcotest.(check int64) "thread tid" 3L ev.thread.tid;
80
+
Alcotest.(check string) "category" "gfx" ev.category;
81
+
Alcotest.(check string) "name" "paint" ev.name;
82
+
match ev.args with
83
+
| [
84
+
("w", `Int64 42L);
85
+
("u", `Unknown 0);
86
+
("s", `String "hi");
87
+
("p", `Pointer 16L);
88
+
("k", `Koid 99L);
89
+
] ->
90
+
()
91
+
| _ ->
92
+
Alcotest.failf "args mismatch: %a" (Fmt.Dump.list Fmt.string)
93
+
(List.map fst ev.args))
94
+
| evs -> Alcotest.failf "expected exactly one event, got %d" (List.length evs)
95
+
96
+
(* --- the four event phases the writer emits --- *)
97
+
98
+
let test_event_phases () =
99
+
let data =
100
+
to_bytes (fun w ->
101
+
Write.counter w ~name:"c" ~thread ~category:"m" ~ts:1L;
102
+
Write.duration_begin w ~name:"span" ~thread ~category:"m" ~ts:2L;
103
+
Write.duration_end w ~name:"span" ~thread ~category:"m" ~ts:3L)
104
+
in
105
+
match List.map (fun (e : Read.Event.t) -> e.ty) (events (parse data)) with
106
+
| [ Counter; Duration_begin; Duration_end ] -> ()
107
+
| tys ->
108
+
Alcotest.failf "expected counter/begin/end, got %d events"
109
+
(List.length tys)
110
+
111
+
(* --- string interning and the 8-byte padding boundary --- *)
112
+
113
+
(* [strlen] rounds a name up to whole 64-bit words; a reader that miscounted the
114
+
padding would recover a truncated or over-long name. Names either side of the
115
+
8-byte boundary must round-trip byte-exact, including the empty name. *)
116
+
let test_string_padding () =
117
+
let names = [ ""; "x"; "1234567"; "12345678"; "123456789" ] in
118
+
let data =
119
+
to_bytes (fun w ->
120
+
List.iteri
121
+
(fun i n ->
122
+
Write.instant_event w ~name:n ~thread ~category:"c"
123
+
~ts:(Int64.of_int i))
124
+
names)
125
+
in
126
+
let got = List.map (fun (e : Read.Event.t) -> e.name) (events (parse data)) in
127
+
Alcotest.(check (list string)) "names round-trip exactly" names got
128
+
129
+
(* --- user object, kernel object, scheduling records --- *)
130
+
131
+
let test_user_object () =
132
+
let data =
133
+
to_bytes (fun w -> Write.user_object w ~name:"cc:root" ~thread 55L)
134
+
in
135
+
match parse data with
136
+
| [ Read.Metadata; Read.User u ] ->
137
+
Alcotest.(check int64) "koid/id" 55L u.id;
138
+
Alcotest.(check string) "name" "cc:root" u.name;
139
+
Alcotest.(check int64) "thread pid" 7L u.thread.pid
140
+
| rs -> Alcotest.failf "expected [Metadata; User], got %d" (List.length rs)
141
+
142
+
let test_kernel_object () =
143
+
let data =
144
+
to_bytes (fun w -> Write.kernel_object w ~name:"main" `Thread 77L)
145
+
in
146
+
match parse data with
147
+
| [ Read.Metadata; Read.Kernel k ] ->
148
+
Alcotest.(check int) "kernel ty (thread=2)" 2 k.ty;
149
+
Alcotest.(check int64) "koid" 77L k.koid;
150
+
Alcotest.(check string) "name" "main" k.name
151
+
| rs -> Alcotest.failf "expected [Metadata; Kernel], got %d" (List.length rs)
152
+
153
+
let test_thread_wakeup () =
154
+
let data = to_bytes (fun w -> Write.thread_wakeup w ~cpu:4 ~ts:900L 12L) in
155
+
match parse data with
156
+
| [ Read.Metadata; Read.Scheduling (Thread_wakeup wk) ] ->
157
+
Alcotest.(check int) "cpu" 4 wk.cpu;
158
+
Alcotest.(check int64) "timestamp" 900L wk.timestamp;
159
+
Alcotest.(check int64) "thread id" 12L wk.id
160
+
| rs ->
161
+
Alcotest.failf "expected [Metadata; Scheduling], got %d" (List.length rs)
162
+
163
+
(* --- hostile: an unrecognised record type is surfaced, not crashed on --- *)
164
+
165
+
(* A single 1-word record of type 5 (which the reader does not model). The low
166
+
nibble is the record type, bits 4-15 the word count: 5 | (1 lsl 4) = 0x15. *)
167
+
let test_unknown_record_type () =
168
+
let buf = Buffer.create 8 in
169
+
le64 buf 0x15L;
170
+
match parse (Buffer.contents buf) with
171
+
| [ Read.Unknown 5 ] -> ()
172
+
| rs -> Alcotest.failf "expected [Unknown 5], got %d records" (List.length rs)
173
+
174
+
(* --- hostile: an event naming an undefined string index --- *)
175
+
176
+
(* A hand-built type-4 event whose name references string index 5, which was
177
+
never defined by a string record. The reader must substitute the placeholder
178
+
rather than raise or read out of bounds.
179
+
180
+
word0 = record-type 4 | words 4 (<<4) | data<<16, where
181
+
data = name-ref 5 (<<32); payload = timestamp, then inline thread pid/tid. *)
182
+
let test_missing_string_index () =
183
+
let buf = Buffer.create 32 in
184
+
le64 buf 0x0005000000000044L;
185
+
le64 buf 42L;
186
+
(* timestamp *)
187
+
le64 buf 7L;
188
+
(* thread pid *)
189
+
le64 buf 0L;
190
+
(* thread tid *)
191
+
match events (parse (Buffer.contents buf)) with
192
+
| [ ev ] ->
193
+
Alcotest.(check string) "placeholder name" "(missing string)" ev.name;
194
+
Alcotest.(check int64) "timestamp" 42L ev.timestamp;
195
+
Alcotest.(check string) "empty category" "" ev.category;
196
+
Alcotest.(check int64) "thread pid" 7L ev.thread.pid
197
+
| evs -> Alcotest.failf "expected one event, got %d" (List.length evs)
198
+
199
+
(* --- hostile: a truncated final record is dropped, not fabricated --- *)
200
+
201
+
let test_truncated_tail () =
202
+
let data =
203
+
to_bytes (fun w ->
204
+
Write.instant_event w ~name:"a" ~thread ~category:"c" ~ts:1L;
205
+
Write.instant_event w ~name:"b" ~thread ~category:"c" ~ts:2L)
206
+
in
207
+
let full = List.length (events (parse data)) in
208
+
Alcotest.(check int) "two events when complete" 2 full;
209
+
(* Drop the last 4 bytes so the final event's payload is incomplete. *)
210
+
let cut = String.sub data 0 (String.length data - 4) in
211
+
Alcotest.(check int)
212
+
"truncated final event is dropped" 1
213
+
(List.length (events (parse cut)))
214
+
215
+
(* --- pretty-printer content --- *)
216
+
217
+
let test_pp_record_content () =
218
+
let data =
219
+
to_bytes (fun w ->
220
+
Write.instant_event w ~name:"render" ~thread ~category:"gfx" ~ts:42L)
221
+
in
222
+
match events (parse data) with
223
+
| [ ev ] ->
224
+
let s = Fmt.str "%a" Read.pp_record (Read.Event ev) in
225
+
let has sub =
226
+
let n = String.length sub and m = String.length s in
227
+
let rec go i =
228
+
if i + n > m then false
229
+
else if String.sub s i n = sub then true
230
+
else go (i + 1)
231
+
in
232
+
n = 0 || go 0
233
+
in
234
+
Alcotest.(check bool) "names the event" true (has "render");
235
+
Alcotest.(check bool) "names the category" true (has "gfx");
236
+
Alcotest.(check bool) "shows the timestamp" true (has "42")
237
+
| _ -> Alcotest.fail "expected one event"
238
+
239
+
let suite =
240
+
( "read",
241
+
[
242
+
Alcotest.test_case "empty input" `Quick test_empty;
243
+
Alcotest.test_case "magic only" `Quick test_magic_only;
244
+
Alcotest.test_case "instant, all arg types" `Quick test_instant_all_args;
245
+
Alcotest.test_case "event phases" `Quick test_event_phases;
246
+
Alcotest.test_case "string padding boundary" `Quick test_string_padding;
247
+
Alcotest.test_case "user object" `Quick test_user_object;
248
+
Alcotest.test_case "kernel object" `Quick test_kernel_object;
249
+
Alcotest.test_case "thread wakeup" `Quick test_thread_wakeup;
250
+
Alcotest.test_case "unknown record type" `Quick test_unknown_record_type;
251
+
Alcotest.test_case "missing string index" `Quick test_missing_string_index;
252
+
Alcotest.test_case "truncated tail dropped" `Quick test_truncated_tail;
253
+
Alcotest.test_case "pp_record content" `Quick test_pp_record_content;
254
+
] )
+4
test/obs/fxt/test_read.mli
+4
test/obs/fxt/test_read.mli
+125
test/obs/fxt/test_write.ml
+125
test/obs/fxt/test_write.ml
···
1
+
(* Tests for {!Observer_fxt.Write}, the Fuchsia trace format writer.
2
+
3
+
Format reference: https://fuchsia.dev/fuchsia-src/reference/tracing/trace-format
4
+
The writer's output is asserted byte-exact where the format fixes it (the
5
+
magic-number record) and confirmed field-by-field by reading it back with
6
+
{!Observer_fxt.Read} otherwise. *)
7
+
8
+
module Write = Observer_fxt.Write
9
+
module Read = Observer_fxt.Read
10
+
11
+
let to_bytes f =
12
+
let buf = Eio.Buf_write.create 0x1000 in
13
+
let w = Write.of_writer buf in
14
+
f w;
15
+
Eio.Buf_write.serialize_to_string buf
16
+
17
+
let parse data =
18
+
match
19
+
Eio.Buf_read.parse_string (fun r -> List.of_seq (Read.records r)) data
20
+
with
21
+
| Ok rs -> rs
22
+
| Error (`Msg m) -> Alcotest.failf "fxt parse: %s" m
23
+
24
+
let events records =
25
+
List.filter_map (function Read.Event e -> Some e | _ -> None) records
26
+
27
+
let thread : Write.thread = { pid = 1L; tid = 2L }
28
+
29
+
(* --- the magic number record is byte-exact --- *)
30
+
31
+
(* Per the Fuchsia trace format, a fresh trace opens with the magic-number
32
+
metadata record 0x0016547846040010 (little-endian on the wire). *)
33
+
let test_magic_bytes () =
34
+
let data = to_bytes (fun _ -> ()) in
35
+
Alcotest.(check int)
36
+
"trace is one 8-byte word (magic only)" 8 (String.length data);
37
+
let word = Bytes.get_int64_le (Bytes.of_string (String.sub data 0 8)) 0 in
38
+
Alcotest.(check int64) "magic number record" 0x0016547846040010L word
39
+
40
+
(* --- the writer is deterministic --- *)
41
+
42
+
let test_deterministic () =
43
+
let emit w =
44
+
Write.duration_begin w ~name:"work" ~thread ~category:"eio" ~ts:10L;
45
+
Write.instant_event w ~name:"tick" ~thread ~category:"eio" ~ts:11L
46
+
~args:[ ("n", `Int64 5L) ];
47
+
Write.duration_end w ~name:"work" ~thread ~category:"eio" ~ts:20L
48
+
in
49
+
Alcotest.(check string)
50
+
"same records -> identical bytes" (to_bytes emit) (to_bytes emit)
51
+
52
+
(* --- strings are interned: a repeated name is written once --- *)
53
+
54
+
(* Two events with the same name share one string record; two with distinct
55
+
names of equal length each carry their own, so the shared trace is strictly
56
+
shorter. This is what proves the string table is used rather than re-inlined. *)
57
+
let test_string_interning () =
58
+
let shared =
59
+
to_bytes (fun w ->
60
+
Write.instant_event w ~name:"shared" ~thread ~category:"c" ~ts:1L;
61
+
Write.instant_event w ~name:"shared" ~thread ~category:"c" ~ts:2L)
62
+
in
63
+
let distinct =
64
+
to_bytes (fun w ->
65
+
Write.instant_event w ~name:"aaaaaa" ~thread ~category:"c" ~ts:1L;
66
+
Write.instant_event w ~name:"bbbbbb" ~thread ~category:"c" ~ts:2L)
67
+
in
68
+
Alcotest.(check bool)
69
+
"interned name yields a shorter trace" true
70
+
(String.length shared < String.length distinct)
71
+
72
+
(* --- a written span pair reads back field-by-field --- *)
73
+
74
+
let test_span_roundtrip () =
75
+
let data =
76
+
to_bytes (fun w ->
77
+
Write.duration_begin w ~name:"compute" ~thread ~category:"eio" ~ts:100L;
78
+
Write.duration_end w ~name:"compute" ~thread ~category:"eio" ~ts:250L)
79
+
in
80
+
match events (parse data) with
81
+
| [ b; e ] ->
82
+
Alcotest.(check bool) "begin phase" true (b.ty = Read.Event.Duration_begin);
83
+
Alcotest.(check string) "begin name" "compute" b.name;
84
+
Alcotest.(check int64) "begin ts" 100L b.timestamp;
85
+
Alcotest.(check bool) "end phase" true (e.ty = Read.Event.Duration_end);
86
+
Alcotest.(check int64) "end ts" 250L e.timestamp;
87
+
Alcotest.(check int64) "thread pid" 1L b.thread.pid;
88
+
Alcotest.(check int64) "thread tid" 2L b.thread.tid
89
+
| evs -> Alcotest.failf "expected two events, got %d" (List.length evs)
90
+
91
+
(* --- an interned thread reference reads back to the same thread --- *)
92
+
93
+
let test_thread_interned () =
94
+
(* Two events on the same thread: the second references the thread record the
95
+
first emitted, and both must resolve to the same (pid, tid). *)
96
+
let data =
97
+
to_bytes (fun w ->
98
+
Write.instant_event w ~name:"a" ~thread ~category:"c" ~ts:1L;
99
+
Write.instant_event w ~name:"b" ~thread ~category:"c" ~ts:2L)
100
+
in
101
+
match events (parse data) with
102
+
| [ e1; e2 ] ->
103
+
Alcotest.(check int64) "e1 pid" 1L e1.thread.pid;
104
+
Alcotest.(check int64) "e2 pid (interned)" 1L e2.thread.pid;
105
+
Alcotest.(check int64) "e2 tid (interned)" 2L e2.thread.tid
106
+
| evs -> Alcotest.failf "expected two events, got %d" (List.length evs)
107
+
108
+
(* --- pp is the opaque writer marker --- *)
109
+
110
+
let test_pp () =
111
+
let buf = Eio.Buf_write.create 16 in
112
+
let w = Write.of_writer buf in
113
+
Alcotest.(check string)
114
+
"writer marker" "<fxt-writer>" (Fmt.str "%a" Write.pp w)
115
+
116
+
let suite =
117
+
( "write",
118
+
[
119
+
Alcotest.test_case "magic number bytes" `Quick test_magic_bytes;
120
+
Alcotest.test_case "deterministic output" `Quick test_deterministic;
121
+
Alcotest.test_case "string interning" `Quick test_string_interning;
122
+
Alcotest.test_case "span round-trip" `Quick test_span_roundtrip;
123
+
Alcotest.test_case "thread interning" `Quick test_thread_interned;
124
+
Alcotest.test_case "pp writer marker" `Quick test_pp;
125
+
] )
+4
test/obs/fxt/test_write.mli
+4
test/obs/fxt/test_write.mli
+17
test/obs/offline/dune
+17
test/obs/offline/dune
+10
test/obs/offline/test.ml
+10
test/obs/offline/test.ml
+69
test/obs/offline/test_catapult_adapter.ml
+69
test/obs/offline/test_catapult_adapter.ml
···
1
+
(* Tests for {!Observer_offline.Catapult_adapter}: the format chosen per file
2
+
extension, a JSON runtime-phase round-trip read back and checked event by
3
+
event, and that opening a sink under a missing directory is rejected. *)
4
+
5
+
module Catapult_adapter = Observer_offline.Catapult_adapter
6
+
module Sink = Catapult_adapter.Sink
7
+
8
+
let format_str : Catapult_adapter.format -> string = function
9
+
| Json -> "json"
10
+
| Protobuf -> "protobuf"
11
+
12
+
(* Only [.json] selects JSON; every other extension (including none and the
13
+
empty name) is protobuf. *)
14
+
let test_format_of_file () =
15
+
let chosen f = format_str (Catapult_adapter.format_of_file f) in
16
+
Alcotest.(check string) ".json is Json" "json" (chosen "x.json");
17
+
Alcotest.(check string) ".pb is Protobuf" "protobuf" (chosen "x.pb");
18
+
Alcotest.(check string) "no extension is Protobuf" "protobuf" (chosen "x");
19
+
Alcotest.(check string) "empty name is Protobuf" "protobuf" (chosen "")
20
+
21
+
(* A GC phase written as begin/end reads back as two Catapult events -- a
22
+
Duration_begin then Duration_end -- carrying the exact microsecond
23
+
timestamps, the phase name, the pid/tid, and the "gc" category. *)
24
+
let test_json_runtime_phase () =
25
+
let path = Filename.temp_file "obs-catapult" ".json" in
26
+
Fun.protect ~finally:(fun () -> if Sys.file_exists path then Sys.remove path)
27
+
@@ fun () ->
28
+
let sink = Sink.v ~format:Json path in
29
+
Sink.runtime_begin sink ~process:7 ~ring:3 ~ts_ns:1_000_000L
30
+
Runtime_events.EV_EXPLICIT_GC_MAJOR;
31
+
Sink.runtime_end sink ~process:7 ~ring:3 ~ts_ns:3_000_000L
32
+
Runtime_events.EV_EXPLICIT_GC_MAJOR;
33
+
Sink.close sink;
34
+
let data = In_channel.with_open_bin path In_channel.input_all in
35
+
let trace = Catapult.of_string_exn data in
36
+
match trace.events with
37
+
| [ (b : Catapult.Event.t); (e : Catapult.Event.t) ] ->
38
+
Alcotest.(check string)
39
+
"begin phase B" "B"
40
+
(Catapult.Phase.to_string b.ph);
41
+
Alcotest.(check (option string))
42
+
"begin name" (Some "explicit_gc_major") b.name;
43
+
Alcotest.(check (float 0.)) "begin ts (us)" 1000. b.ts;
44
+
Alcotest.(check (option int)) "begin pid" (Some 7) b.pid;
45
+
Alcotest.(check (option int)) "begin tid" (Some 3) b.tid;
46
+
Alcotest.(check (list string)) "begin category" [ "gc" ] b.cat;
47
+
Alcotest.(check string) "end phase E" "E" (Catapult.Phase.to_string e.ph);
48
+
Alcotest.(check (option string))
49
+
"end name" (Some "explicit_gc_major") e.name;
50
+
Alcotest.(check (float 0.)) "end ts (us)" 3000. e.ts;
51
+
Alcotest.(check (option int)) "end pid" (Some 7) e.pid;
52
+
Alcotest.(check (option int)) "end tid" (Some 3) e.tid
53
+
| l -> Alcotest.failf "expected exactly 2 events, got %d" (List.length l)
54
+
55
+
(* Hostile: opening a sink under a directory that does not exist raises
56
+
[Sys_error] from the underlying open. *)
57
+
let test_missing_dir_raises () =
58
+
match Sink.v ~format:Json "/nonexistent-obs-dir/out.json" with
59
+
| _ -> Alcotest.fail "opening a sink under a missing directory must raise"
60
+
| exception Sys_error _ -> ()
61
+
62
+
let suite =
63
+
( "catapult_adapter",
64
+
[
65
+
Alcotest.test_case "format of file" `Quick test_format_of_file;
66
+
Alcotest.test_case "json runtime phase" `Quick test_json_runtime_phase;
67
+
Alcotest.test_case "missing directory raises" `Quick
68
+
test_missing_dir_raises;
69
+
] )
+4
test/obs/offline/test_catapult_adapter.mli
+4
test/obs/offline/test_catapult_adapter.mli
+126
test/obs/offline/test_collector.ml
+126
test/obs/offline/test_collector.ml
···
1
+
(* Tests for {!Observer_offline.Collector}: producer registration, clock-domain
2
+
placement and rebasing, and memtrace CTF reassembly. All of [Collector]'s API
3
+
is synchronous, so these run without an Eio scheduler. *)
4
+
5
+
module Frame = Observe_protocol.Frame
6
+
module Collector = Observer_offline.Collector
7
+
module Fxt_adapter = Observer_offline.Fxt_adapter
8
+
9
+
let hello os_pid =
10
+
{
11
+
Frame.producer_name = "test";
12
+
kind = Host;
13
+
os_pid;
14
+
word_size = 8;
15
+
clock_unix_ns = 0L;
16
+
clock_mono_ns = 0L;
17
+
trace_id = "";
18
+
root_span_id = 0L;
19
+
parent_span_id = 0L;
20
+
}
21
+
22
+
let sink () = Fxt_adapter.Sink.v ()
23
+
24
+
(* Producer ids are collector-scoped: two producers that share an OS pid still
25
+
get distinct ids, and [count] tracks how many registered. *)
26
+
let test_scoped_ids () =
27
+
let c = Collector.v (sink ()) in
28
+
let id0 = Collector.register c (hello 999) ~mono_ns:0L in
29
+
let id1 = Collector.register c (hello 999) ~mono_ns:0L in
30
+
Alcotest.(check int) "first id" 0 id0;
31
+
Alcotest.(check int) "second id, same pid" 1 id1;
32
+
Alcotest.(check int) "count" 2 (Collector.count c);
33
+
Alcotest.(check int)
34
+
"two producers listed" 2
35
+
(List.length (Collector.producers c))
36
+
37
+
(* A producer whose monotonic sample is within the same-host threshold shares
38
+
the collector's clock domain and joins exactly; one 10s off is a foreign
39
+
domain, making the collection approximate. *)
40
+
let test_clock_domains () =
41
+
let c = Collector.v (sink ()) in
42
+
let _ = Collector.register c (hello 1) ~mono_ns:2_000_000L in
43
+
Alcotest.(check bool) "same-host stays exact" false (Collector.approximate c);
44
+
Alcotest.(check int) "one clock domain" 1 (Collector.clock_domain_count c);
45
+
let _ = Collector.register c (hello 2) ~mono_ns:10_000_000_000L in
46
+
Alcotest.(check bool)
47
+
"foreign domain is approximate" true (Collector.approximate c);
48
+
Alcotest.(check int) "two clock domains" 2 (Collector.clock_domain_count c)
49
+
50
+
(* A foreign producer's event at time T, rebased by its offset, lands exactly
51
+
where a same-host producer's event at T+offset lands: proven by the two sinks
52
+
emitting byte-identical fxt. *)
53
+
let test_clock_rebasing () =
54
+
let offset = 10_000_000_000L in
55
+
let gc ts = Frame.Gc_begin { ring = 0; ts_ns = ts; phase = "minor" } in
56
+
let fxt ~mono_ns ~event_ts =
57
+
let s = sink () in
58
+
let c = Collector.v s in
59
+
let id = Collector.register c (hello 1) ~mono_ns in
60
+
Collector.event c ~producer_id:id (gc event_ts);
61
+
Fxt_adapter.Sink.to_string s
62
+
in
63
+
let foreign = fxt ~mono_ns:offset ~event_ts:5L in
64
+
let local = fxt ~mono_ns:0L ~event_ts:(Int64.add 5L offset) in
65
+
Alcotest.(check string) "rebased onto the collector timeline" local foreign
66
+
67
+
(* Memtrace chunks are opaque CTF slices; the collector concatenates a
68
+
producer's chunks back into its byte-exact stream, keyed by producer id. *)
69
+
let test_ctf_reassembly () =
70
+
let c = Collector.v (sink ()) in
71
+
let id = Collector.register c (hello 1) ~mono_ns:0L in
72
+
List.iter
73
+
(fun s -> Collector.event c ~producer_id:id (Frame.Memtrace_chunk s))
74
+
[ "abc"; "def"; "ghi" ];
75
+
match Collector.ctf c with
76
+
| [ (pid, bytes) ] ->
77
+
Alcotest.(check int) "keyed by producer id" id pid;
78
+
Alcotest.(check string) "chunks in order" "abcdefghi" bytes
79
+
| l -> Alcotest.failf "expected one ctf stream, got %d" (List.length l)
80
+
81
+
(* Only producers that actually sent chunks appear in [ctf]: a producer with no
82
+
Memtrace_chunk contributes no (empty) entry. *)
83
+
let test_ctf_only_nonempty () =
84
+
let c = Collector.v (sink ()) in
85
+
let id0 = Collector.register c (hello 1) ~mono_ns:0L in
86
+
let _id1 = Collector.register c (hello 2) ~mono_ns:0L in
87
+
Collector.event c ~producer_id:id0 (Frame.Memtrace_chunk "data");
88
+
match Collector.ctf c with
89
+
| [ (pid, bytes) ] ->
90
+
Alcotest.(check int) "only the chunk-sending producer" id0 pid;
91
+
Alcotest.(check string) "its bytes" "data" bytes
92
+
| l ->
93
+
Alcotest.failf "expected exactly one ctf stream, got %d" (List.length l)
94
+
95
+
(* Hello/Bye/Unknown routed through [event] (rather than [register]) are no-ops:
96
+
they neither register a producer nor touch the CTF table. *)
97
+
let test_control_frames_are_noops () =
98
+
let c = Collector.v (sink ()) in
99
+
let id = Collector.register c (hello 1) ~mono_ns:0L in
100
+
Collector.event c ~producer_id:id (Frame.Hello (hello 2));
101
+
Collector.event c ~producer_id:id Frame.Bye;
102
+
Collector.event c ~producer_id:id Frame.Unknown;
103
+
Alcotest.(check int) "no new producers" 1 (Collector.count c);
104
+
Alcotest.(check int) "no ctf recorded" 0 (List.length (Collector.ctf c))
105
+
106
+
(* An event for a producer id that was never registered has no clock offset (so
107
+
it is fed at offset 0) and must not raise or register anything. *)
108
+
let test_unregistered_producer () =
109
+
let c = Collector.v (sink ()) in
110
+
Collector.event c ~producer_id:99
111
+
(Frame.Gc_begin { ring = 0; ts_ns = 1L; phase = "minor" });
112
+
Alcotest.(check int) "still no producers registered" 0 (Collector.count c)
113
+
114
+
let suite =
115
+
( "collector",
116
+
[
117
+
Alcotest.test_case "collector-scoped ids" `Quick test_scoped_ids;
118
+
Alcotest.test_case "clock domains" `Quick test_clock_domains;
119
+
Alcotest.test_case "clock rebasing" `Quick test_clock_rebasing;
120
+
Alcotest.test_case "ctf reassembly" `Quick test_ctf_reassembly;
121
+
Alcotest.test_case "ctf only non-empty" `Quick test_ctf_only_nonempty;
122
+
Alcotest.test_case "control frames are no-ops" `Quick
123
+
test_control_frames_are_noops;
124
+
Alcotest.test_case "unregistered producer" `Quick
125
+
test_unregistered_producer;
126
+
] )
+4
test/obs/offline/test_collector.mli
+4
test/obs/offline/test_collector.mli
+117
test/obs/offline/test_collector_server.ml
+117
test/obs/offline/test_collector_server.ml
···
1
+
(* Tests for {!Observer_offline.Collector_server}: a producer connecting over an
2
+
AF_UNIX socket and sending Hello/Gc_begin/Bye is registered in the collector's
3
+
clock domain, a stream that is not a valid frame is dropped without a
4
+
producer, and vsock availability is answered per platform.
5
+
6
+
Frames are written by a plain {!Eio.Net} client rather than a producer
7
+
forwarder: the forwarder's [stop] joins a background domain and would deadlock
8
+
against the handler fiber. The client socket is closed explicitly because
9
+
[serve] uses [accept_fork ~sw], so the handler fiber ends only on EOF. *)
10
+
11
+
module Collector_server = Observer_offline.Collector_server
12
+
module Frame = Observe_protocol.Frame
13
+
module Fxt_sink = Observer_offline.Fxt_adapter.Sink
14
+
15
+
let base_hello =
16
+
{
17
+
Frame.producer_name = "producer-under-test";
18
+
kind = Host;
19
+
os_pid = 4242;
20
+
word_size = 8;
21
+
clock_unix_ns = 0L;
22
+
clock_mono_ns = 0L;
23
+
trace_id = "";
24
+
root_span_id = 0L;
25
+
parent_span_id = 0L;
26
+
}
27
+
28
+
let mono_ns mono = Mtime.to_uint64_ns (Eio.Time.Mono.now mono)
29
+
let server () = Collector_server.v (Fxt_sink.v ())
30
+
31
+
(* Poll [cond], yielding to the collector's handler fibers between checks.
32
+
Bounded, so a wedged handler fails the test loudly instead of hanging. *)
33
+
let wait_until env cond =
34
+
let clock = Eio.Stdenv.clock env in
35
+
let rec go n =
36
+
if cond () then ()
37
+
else if n <= 0 then Alcotest.fail "timed out waiting for collector state"
38
+
else (
39
+
Eio.Time.sleep clock 0.005;
40
+
go (n - 1))
41
+
in
42
+
go 400
43
+
44
+
let test_hello_gc_bye () =
45
+
Eio_main.run @@ fun env ->
46
+
Eio.Switch.run @@ fun sw ->
47
+
let net = Eio.Stdenv.net env in
48
+
let mono = Eio.Stdenv.mono_clock env in
49
+
let t = server () in
50
+
let path = "obs-collector-pos.sock" in
51
+
Collector_server.serve ~sw ~net ~mono t ~path;
52
+
let flow = Eio.Net.connect ~sw net (`Unix path) in
53
+
(* Sample the same monotonic clock the collector reads at accept, so the
54
+
producer lands in domain 0 (offset well under the same-host threshold). *)
55
+
let hello = { base_hello with clock_mono_ns = mono_ns mono } in
56
+
let buf = Buffer.create 256 in
57
+
Frame.add_framed buf (Frame.Hello hello);
58
+
Frame.add_framed buf
59
+
(Frame.Gc_begin { ring = 0; ts_ns = 1L; phase = "minor" });
60
+
Frame.add_framed buf Frame.Bye;
61
+
Eio.Flow.copy_string (Buffer.contents buf) flow;
62
+
wait_until env (fun () -> Collector_server.producer_count t = 1);
63
+
Alcotest.(check int)
64
+
"one producer connected" 1
65
+
(Collector_server.producer_count t);
66
+
Alcotest.(check int)
67
+
"single clock domain" 1
68
+
(Collector_server.clock_domain_count t);
69
+
(match Collector_server.producers t with
70
+
| [ (id, h) ] ->
71
+
Alcotest.(check int) "producer id is 0" 0 id;
72
+
Alcotest.(check string)
73
+
"hello producer name reported" base_hello.Frame.producer_name
74
+
h.Frame.producer_name;
75
+
Alcotest.(check int)
76
+
"hello os pid reported" base_hello.Frame.os_pid h.Frame.os_pid
77
+
| l -> Alcotest.failf "expected one producer, got %d" (List.length l));
78
+
Eio.Flow.close flow;
79
+
wait_until env (fun () -> Collector_server.active_connections t = 0)
80
+
81
+
let test_malformed_dropped () =
82
+
Eio_main.run @@ fun env ->
83
+
Eio.Switch.run @@ fun sw ->
84
+
let net = Eio.Stdenv.net env in
85
+
let mono = Eio.Stdenv.mono_clock env in
86
+
let t = server () in
87
+
let path = "obs-collector-neg.sock" in
88
+
Collector_server.serve ~sw ~net ~mono t ~path;
89
+
let flow = Eio.Net.connect ~sw net (`Unix path) in
90
+
(* A leading byte that is not the frame tag makes [Frame.next] fail, so the
91
+
handler drops the stream rather than registering a producer. *)
92
+
Eio.Flow.copy_string "not-a-valid-frame" flow;
93
+
wait_until env (fun () -> Collector_server.active_connections t = 1);
94
+
Alcotest.(check int)
95
+
"no producer from garbage" 0
96
+
(Collector_server.producer_count t);
97
+
Eio.Flow.close flow;
98
+
wait_until env (fun () -> Collector_server.active_connections t = 0);
99
+
Alcotest.(check int)
100
+
"still no producer after drop" 0
101
+
(Collector_server.producer_count t)
102
+
103
+
(* [vsock_available] answers per platform without raising: false on macOS (no
104
+
AF_VSOCK), true on Linux. *)
105
+
let test_vsock_available () =
106
+
let is_linux = Sys.file_exists "/proc/version" in
107
+
Alcotest.(check bool)
108
+
"vsock availability matches platform" is_linux
109
+
(Collector_server.vsock_available ())
110
+
111
+
let suite =
112
+
( "collector_server",
113
+
[
114
+
Alcotest.test_case "hello gc bye" `Quick test_hello_gc_bye;
115
+
Alcotest.test_case "malformed dropped" `Quick test_malformed_dropped;
116
+
Alcotest.test_case "vsock available" `Quick test_vsock_available;
117
+
] )
+4
test/obs/offline/test_collector_server.mli
+4
test/obs/offline/test_collector_server.mli
+137
test/obs/offline/test_ctf_input.ml
+137
test/obs/offline/test_ctf_input.ml
···
1
+
(* Tests for {!Observer_offline.Ctf_input}: reading a memtrace [.ctf] allocation
2
+
trace into the shared {!Observer_model.Model}. A deterministic trace authored
3
+
with {!Trace.Writer} lets the test pin exact allocation words and the resolved
4
+
backtrace site, and check that a missing or malformed trace is rejected loudly
5
+
rather than silently ingesting nothing. *)
6
+
7
+
module Ctf_input = Observer_offline.Ctf_input
8
+
module Model = Observer_model.Model
9
+
module Report = Observer_model.Report
10
+
module Trace = Observe_memtrace.Memtrace.Trace
11
+
module Event = Trace.Event
12
+
module Location = Trace.Location
13
+
14
+
let sample_rate = 1.0
15
+
let alloc_words = 100
16
+
let start_us = 1_000_000L
17
+
18
+
(* The one location the sole allocation is attributed to. *)
19
+
let site_loc : Location.t =
20
+
{
21
+
filename = "alloc_site.ml";
22
+
line = 77;
23
+
start_char = 0;
24
+
end_char = 10;
25
+
defname = "Obs_test.hot_alloc";
26
+
}
27
+
28
+
(* The site string [Ctf_input.pp_site] renders from [site_loc]:
29
+
"defname (basename:line)". *)
30
+
let site_str = "Obs_test.hot_alloc (alloc_site.ml:77)"
31
+
32
+
let info : Trace.Info.t =
33
+
{
34
+
sample_rate;
35
+
word_size = 64;
36
+
executable_name = "obs-test";
37
+
host_name = "obs-host";
38
+
ocaml_runtime_params = "";
39
+
pid = 4242L;
40
+
initial_domain = Trace.Domain_id.Expert.of_int 0;
41
+
start_time = Trace.Timestamp.of_int64 start_us;
42
+
context = None;
43
+
}
44
+
45
+
let loc_code = Trace.Location_code.Expert.of_int 1
46
+
47
+
(* Author a real single-allocation [.ctf] at [path]: [alloc_words] words in one
48
+
minor allocation whose innermost (and only) backtrace frame is [site_loc]. *)
49
+
let write_trace path =
50
+
let fd =
51
+
Unix.openfile path [ Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC ] 0o600
52
+
in
53
+
let w = Trace.Writer.Multiplexed_domains.v fd info in
54
+
let decode code =
55
+
if (code : Trace.Location_code.t :> int) = 1 then [ site_loc ]
56
+
else [ Location.unknown ]
57
+
in
58
+
let ev : Event.t =
59
+
Alloc
60
+
{
61
+
obj_id = Trace.Obj_id.Expert.of_int 0;
62
+
length = alloc_words;
63
+
domain = Trace.Domain_id.Expert.of_int 0;
64
+
nsamples = 1;
65
+
source = Trace.Allocation_source.Minor;
66
+
backtrace_buffer = [| loc_code |];
67
+
backtrace_length = 1;
68
+
common_prefix = 0;
69
+
span_name = None;
70
+
}
71
+
in
72
+
let ts = Trace.Timestamp.of_int64 (Int64.add start_us 1000L) in
73
+
Trace.Writer.Multiplexed_domains.put_event w ~decode_callstack_entry:decode ts
74
+
ev;
75
+
Trace.Writer.Multiplexed_domains.flush w;
76
+
Unix.close fd
77
+
78
+
let with_temp suffix f =
79
+
let path = Filename.temp_file "obs-ctf" suffix in
80
+
Fun.protect
81
+
~finally:(fun () -> if Sys.file_exists path then Sys.remove path)
82
+
(fun () -> f path)
83
+
84
+
(* Feeding a real trace records the exact sample rate, attributes the allocation
85
+
to its site with exact words, and marks the model allocation-only. *)
86
+
let test_feed_allocation () =
87
+
with_temp ".ctf" @@ fun path ->
88
+
write_trace path;
89
+
(* The rate [feed] forwards to the model is the one carried by the trace. *)
90
+
let r = Trace.Reader.open_ ~filename:path in
91
+
Alcotest.(check (float 0.))
92
+
"authored sample rate (forwarded to the model)" sample_rate
93
+
(Trace.Reader.info r).sample_rate;
94
+
Trace.Reader.close r;
95
+
let model = Model.v () in
96
+
Ctf_input.feed model path;
97
+
let rep : Report.t = Model.report model in
98
+
Alcotest.(check int) "allocation words" alloc_words rep.total_alloc_words;
99
+
Alcotest.(check (option string)) "ctf path recorded" (Some path) rep.ctf_path;
100
+
Alcotest.(check bool) "allocation-only trace" true rep.allocation_only;
101
+
match rep.top_sites with
102
+
| [ s ] ->
103
+
Alcotest.(check string) "resolved allocation site" site_str s.site;
104
+
Alcotest.(check int) "site words" alloc_words s.words;
105
+
Alcotest.(check (float 0.)) "site share of total" 1.0 s.share
106
+
| l -> Alcotest.failf "expected exactly one top site, got %d" (List.length l)
107
+
108
+
(* Hostile: a path that cannot be opened raises [Sys_error] naming the path (the
109
+
documented contract), not a leaked [Unix_error]. *)
110
+
let test_missing_raises () =
111
+
let missing = "/nonexistent/obs-x.ctf" in
112
+
match Ctf_input.feed (Model.v ()) missing with
113
+
| () -> Alcotest.fail "feed of a missing .ctf must raise, not return"
114
+
| exception Sys_error msg ->
115
+
Alcotest.(check bool)
116
+
"Sys_error names the ctf path" true
117
+
(String.starts_with ~prefix:(missing ^ ":") msg)
118
+
119
+
(* Hostile: a file of garbage bytes is not a CTF trace, so [feed] raises the
120
+
memtrace parse error rather than silently feeding nothing to the model. *)
121
+
let test_garbage_raises () =
122
+
with_temp ".ctf" @@ fun path ->
123
+
let oc = open_out_bin path in
124
+
output_string oc (String.make 4096 '\xff');
125
+
close_out oc;
126
+
match Ctf_input.feed (Model.v ()) path with
127
+
| () -> Alcotest.fail "feed of a garbage .ctf must raise, not silently ingest"
128
+
| exception Observe_memtrace.Memtrace.Trace.Parse_error _ -> ()
129
+
130
+
let suite =
131
+
( "ctf_input",
132
+
[
133
+
Alcotest.test_case "feed allocation" `Quick test_feed_allocation;
134
+
Alcotest.test_case "missing file raises Sys_error" `Quick
135
+
test_missing_raises;
136
+
Alcotest.test_case "garbage file raises" `Quick test_garbage_raises;
137
+
] )
+4
test/obs/offline/test_ctf_input.mli
+4
test/obs/offline/test_ctf_input.mli
+112
test/obs/offline/test_fxt_adapter.ml
+112
test/obs/offline/test_fxt_adapter.ml
···
1
+
(* Tests for {!Observer_offline.Fxt_adapter}: the writer ({!Sink}) emits one fxt
2
+
record per runtime_events record and the reader ({!feed_string}) parses it
3
+
back into the shared model. Driving both directions and reading exact numbers
4
+
out of the model proves the pair is a true inverse. *)
5
+
6
+
module Sink = Observer_offline.Fxt_adapter.Sink
7
+
module Fxt_adapter = Observer_offline.Fxt_adapter
8
+
module Model = Observer_model.Model
9
+
module Rep = Observer_model.Report
10
+
11
+
let ms n = Int64.of_float (n *. 1e6)
12
+
13
+
let roundtrip build =
14
+
let sink = Sink.v () in
15
+
build sink;
16
+
let model = Model.v () in
17
+
Fxt_adapter.feed_string model (Sink.to_string sink);
18
+
Model.report model
19
+
20
+
(* A span written as a duration begin/end pair reads back with its exact
21
+
inclusive time: 0..100ms -> 0.100s. *)
22
+
let test_span_roundtrip () =
23
+
let r =
24
+
roundtrip (fun s ->
25
+
Sink.eio_event s ~ts_ns:(ms 0.) (`Enter_span "work");
26
+
Sink.eio_event s ~ts_ns:(ms 100.) `Exit_span)
27
+
in
28
+
let w = List.find (fun (sp : Rep.span) -> sp.name = "work") r.spans in
29
+
Alcotest.(check (float 1e-9)) "inclusive time" 0.100 w.incl
30
+
31
+
(* The GC allocation counters carry their units through the fxt round-trip:
32
+
[minor-allocated]/[minor-promoted] are bytes (8 per word), [major-allocated]
33
+
is words. Two 800-byte minor samples make 200 words. *)
34
+
let test_gc_counters () =
35
+
let r =
36
+
roundtrip (fun s ->
37
+
Sink.gc_counter s ~ts_ns:(ms 1.) ~name:"minor-allocated" 800;
38
+
Sink.gc_counter s ~ts_ns:(ms 2.) ~name:"minor-allocated" 800;
39
+
Sink.gc_counter s ~ts_ns:(ms 3.) ~name:"major-allocated" 64;
40
+
Sink.gc_counter s ~ts_ns:(ms 4.) ~name:"minor-promoted" 80)
41
+
in
42
+
Alcotest.(check int) "minor words (1600B / 8)" 200 r.minor_words;
43
+
Alcotest.(check int) "major words" 64 r.major_words;
44
+
Alcotest.(check int) "promoted words (80B / 8)" 10 r.promoted_words
45
+
46
+
(* GC phase spans read back into collection counts and timing: a 2ms minor
47
+
phase is one minor collection, and gc_time reflects its 2ms. *)
48
+
let test_gc_phase () =
49
+
let r =
50
+
roundtrip (fun s ->
51
+
Sink.gc_begin s ~ts_ns:(ms 0.) Runtime_events.EV_MINOR;
52
+
Sink.gc_end s ~ts_ns:(ms 2.) Runtime_events.EV_MINOR)
53
+
in
54
+
Alcotest.(check int) "one minor collection" 1 r.minor_collections;
55
+
Alcotest.(check (float 1e-9)) "gc time" 0.002 r.gc_time
56
+
57
+
(* The thread of every record is (pid=process, tid=ring), so records written for
58
+
two distinct processes never merge: two same-id fibers stay two fibers. *)
59
+
let test_process_isolation () =
60
+
let r =
61
+
roundtrip (fun s ->
62
+
Sink.eio_event s ~process:0 ~ts_ns:(ms 0.) (`Fiber 1);
63
+
Sink.eio_event s ~process:0 ~ts_ns:(ms 10.) (`Exit_fiber 1);
64
+
Sink.eio_event s ~process:1 ~ts_ns:(ms 0.) (`Fiber 1);
65
+
Sink.eio_event s ~process:1 ~ts_ns:(ms 20.) (`Exit_fiber 1))
66
+
in
67
+
Alcotest.(check int) "two fibers, not merged" 2 r.n_fibers
68
+
69
+
(* The cancellation-context kind is encoded in the user-object name ("cc:switch")
70
+
and recovered by the reader, so the model still knows a fiber's spawn context
71
+
after the fxt round-trip. *)
72
+
let test_cc_kind_survives () =
73
+
let r =
74
+
roundtrip (fun s ->
75
+
Sink.eio_event s ~ts_ns:(ms 0.) (`Create (1, `Cc Switch));
76
+
Sink.eio_event s ~ts_ns:(ms 1.) (`Create (2, `Fiber_in 1));
77
+
Sink.eio_event s ~ts_ns:(ms 1.) (`Fiber 2);
78
+
Sink.eio_event s ~ts_ns:(ms 5.) (`Exit_fiber 2))
79
+
in
80
+
Alcotest.(check int) "one cancellation context" 1 r.n_ccs;
81
+
let f = List.find (fun (f : Rep.fiber) -> f.id = 2) r.fibers in
82
+
Alcotest.(check string) "spawn context recovered" "cc:switch" f.spawn
83
+
84
+
(* Hostile: a malformed trace (a record claiming a negative payload) must be
85
+
rejected loudly, not silently absorbed into the model. *)
86
+
let test_malformed_rejected () =
87
+
let le64 buf x =
88
+
let b = Bytes.create 8 in
89
+
Bytes.set_int64_le b 0 x;
90
+
Buffer.add_bytes buf b
91
+
in
92
+
let buf = Buffer.create 16 in
93
+
le64 buf 0x0016547846040010L (* magic *);
94
+
le64 buf 0x05L (* type 5, word-count 0 -> payload size -8 *);
95
+
let raised =
96
+
try
97
+
Fxt_adapter.feed_string (Model.v ()) (Buffer.contents buf);
98
+
false
99
+
with Failure _ | Invalid_argument _ -> true
100
+
in
101
+
Alcotest.(check bool) "malformed fxt rejected" true raised
102
+
103
+
let suite =
104
+
( "fxt_adapter",
105
+
[
106
+
Alcotest.test_case "span round-trip" `Quick test_span_roundtrip;
107
+
Alcotest.test_case "gc counters" `Quick test_gc_counters;
108
+
Alcotest.test_case "gc phase" `Quick test_gc_phase;
109
+
Alcotest.test_case "process isolation" `Quick test_process_isolation;
110
+
Alcotest.test_case "cc kind survives" `Quick test_cc_kind_survives;
111
+
Alcotest.test_case "malformed rejected" `Quick test_malformed_rejected;
112
+
] )
+4
test/obs/offline/test_fxt_adapter.mli
+4
test/obs/offline/test_fxt_adapter.mli
+63
test/obs/offline/test_sqlite_adapter.ml
+63
test/obs/offline/test_sqlite_adapter.ml
···
1
+
(* Tests for {!Observer_offline.Sqlite_adapter}: a GC-phase begin/end pair lands
2
+
as one [slice] row whose duration is exactly end-minus-begin, a registered
3
+
process keeps its pid and name, and a [gc_end] with no open begin neither
4
+
raises nor writes a slice. The database is reopened with the [sqlite] library
5
+
and its rows asserted directly. *)
6
+
7
+
module Sink = Observer_offline.Sqlite_adapter.Sink
8
+
9
+
let rows db sql = List.of_seq (Sqlite.Raw.query db sql ~params:[])
10
+
11
+
let with_db f =
12
+
Eio_main.run @@ fun env ->
13
+
let fs = Eio.Stdenv.fs env in
14
+
Eio.Switch.run @@ fun sw ->
15
+
let path = Filename.temp_file "obs-sqlite" ".db" in
16
+
Fun.protect
17
+
~finally:(fun () -> if Sys.file_exists path then Sys.remove path)
18
+
(fun () -> f ~fs ~sw path)
19
+
20
+
(* Reopen the database written by [Sink] and hand [f] a read handle. *)
21
+
let reopen ~fs ~sw path f =
22
+
let db = Sqlite.open_ ~sw ~create:false Eio.Path.(fs / path) in
23
+
Fun.protect ~finally:(fun () -> Sqlite.close db) (fun () -> f db)
24
+
25
+
let test_slice_and_process () =
26
+
with_db @@ fun ~fs ~sw path ->
27
+
let sink = Sink.v ~fs ~sw path in
28
+
Sink.register_process sink ~upid:1 ~pid:4242 ~name:"myproc";
29
+
Sink.gc_begin sink ~upid:1 ~ring:0 ~ts_ns:1000L ~phase:"minor";
30
+
Sink.gc_end sink ~upid:1 ~ring:0 ~ts_ns:1500L;
31
+
Sink.close sink;
32
+
reopen ~fs ~sw path @@ fun db ->
33
+
(match rows db "SELECT ts, dur, name FROM slice" with
34
+
| [ [ Sqlite.Int ts; Sqlite.Int dur; Sqlite.Text name ] ] ->
35
+
Alcotest.(check int64) "slice ts is the begin timestamp" 1000L ts;
36
+
Alcotest.(check int64) "slice dur is end minus begin" 500L dur;
37
+
Alcotest.(check string) "slice name is the phase" "minor" name
38
+
| l -> Alcotest.failf "expected exactly one slice row, got %d" (List.length l));
39
+
match rows db "SELECT pid, name FROM process" with
40
+
| [ [ Sqlite.Int pid; Sqlite.Text name ] ] ->
41
+
Alcotest.(check int64) "process pid" 4242L pid;
42
+
Alcotest.(check string) "process name" "myproc" name
43
+
| l ->
44
+
Alcotest.failf "expected exactly one process row, got %d" (List.length l)
45
+
46
+
(* Hostile: a [gc_end] with no matching [gc_begin] must not raise (reaching the
47
+
assertion proves it) and must leave the [slice] table empty. *)
48
+
let test_unmatched_gc_end () =
49
+
with_db @@ fun ~fs ~sw path ->
50
+
let sink = Sink.v ~fs ~sw path in
51
+
Sink.gc_end sink ~upid:1 ~ring:0 ~ts_ns:1500L;
52
+
Sink.close sink;
53
+
reopen ~fs ~sw path @@ fun db ->
54
+
match rows db "SELECT count(*) FROM slice" with
55
+
| [ [ Sqlite.Int n ] ] -> Alcotest.(check int64) "no slice row created" 0L n
56
+
| l -> Alcotest.failf "unexpected count result: %d rows" (List.length l)
57
+
58
+
let suite =
59
+
( "sqlite_adapter",
60
+
[
61
+
Alcotest.test_case "slice and process" `Quick test_slice_and_process;
62
+
Alcotest.test_case "unmatched gc_end" `Quick test_unmatched_gc_end;
63
+
] )
+4
test/obs/offline/test_sqlite_adapter.mli
+4
test/obs/offline/test_sqlite_adapter.mli
+1
-1
test/obs_eio/dune
+1
-1
test/obs_eio/dune
+1
test/producer/dune
+1
test/producer/dune
+1
test/protocol/dune
+1
test/protocol/dune