Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299//! Honest process measurements for the Grafana runtime row.//!//! The upstream dashboard gets these from the Go Prometheus collector. Stream//! measures the same process boundaries directly instead of publishing//! synthetic `go_*` series. Linux memory/thread values come from procfs;//! Darwin uses the kernel's proc_pidinfo task snapshot. CPU is cumulative//! getrusage time on both platforms, and FD counts enumerate this process's//! descriptor directory.
const std = @import("std");
const Io = std.Io;
pub const Snapshot = struct { cpu_us: ?u64 = null, resident_memory_bytes: ?u64 = null, virtual_memory_bytes: ?u64 = null, threads: ?u64 = null, open_fds: ?u64 = null, max_fds: ?u64 = null, minor_page_faults: ?u64 = null, major_page_faults: ?u64 = null, voluntary_context_switches: ?u64 = null, involuntary_context_switches: ?u64 = null, malloc_system_bytes: ?u64 = null, malloc_free_total_bytes: ?u64 = null, malloc_arena_heaps: ?u64 = null,};
pub fn collect(io: Io) Snapshot { var result: Snapshot = .{}; if (readMallocInfo()) |info| { result.malloc_system_bytes = info.system_bytes; result.malloc_free_total_bytes = info.free_bytes; result.malloc_arena_heaps = info.heap_count; }
const usage = std.posix.getrusage(std.posix.rusage.SELF); result.cpu_us = timevalMicros(usage.utime) +| timevalMicros(usage.stime); result.minor_page_faults = nonNegative(usage.minflt); result.major_page_faults = nonNegative(usage.majflt); result.voluntary_context_switches = nonNegative(usage.nvcsw); result.involuntary_context_switches = nonNegative(usage.nivcsw);
const limits = std.posix.getrlimit(.NOFILE) catch null; if (limits) |value| { // The soft limit is the process's effective maximum. The hard limit // is merely the ceiling to which a privileged process may raise it. if (value.cur != std.math.maxInt(std.posix.rlim_t)) result.max_fds = @intCast(value.cur); } result.open_fds = countOpenFds(io);
switch (@import("builtin").os.tag) { .linux => { var buf: [16 * 1024]u8 = undefined; const status = Io.Dir.cwd().readFile(io, "/proc/self/status", &buf) catch return result; if (status.len == buf.len) return result; const parsed = parseLinuxStatus(status); result.resident_memory_bytes = parsed.resident_memory_bytes; result.virtual_memory_bytes = parsed.virtual_memory_bytes; result.threads = parsed.threads; }, .macos => { if (darwinTaskInfo()) |task| { result.resident_memory_bytes = task.resident_size; result.virtual_memory_bytes = task.virtual_size; if (task.thread_count >= 0) result.threads = @intCast(task.thread_count); } }, else => {}, } return result;}
fn timevalMicros(value: anytype) u64 { if (value.sec < 0 or value.usec < 0) return 0; return @as(u64, @intCast(value.sec)) *| std.time.us_per_s +| @as(u64, @intCast(value.usec));}
fn nonNegative(value: anytype) ?u64 { if (value < 0) return null; return @intCast(value);}
fn countOpenFds(io: Io) ?u64 { const path = switch (@import("builtin").os.tag) { .linux => "/proc/self/fd", .macos => "/dev/fd", else => return null, }; var dir = Io.Dir.cwd().openDir(io, path, .{ .iterate = true }) catch return null; defer dir.close(io); var iterator = dir.iterate(); var count: u64 = 0; while (iterator.next(io) catch return null) |_| count += 1; // Opening the descriptor directory itself adds one descriptor to the // listing. Remove that observation artifact from the process count. return count -| 1;}
const LinuxStatus = struct { resident_memory_bytes: ?u64 = null, virtual_memory_bytes: ?u64 = null, threads: ?u64 = null,};
fn parseLinuxStatus(bytes: []const u8) LinuxStatus { var result: LinuxStatus = .{}; var lines = std.mem.splitScalar(u8, bytes, '\n'); while (lines.next()) |line| { if (parseKbField(line, "VmRSS:")) |value| result.resident_memory_bytes = value; if (parseKbField(line, "VmSize:")) |value| result.virtual_memory_bytes = value; if (parseIntegerField(line, "Threads:")) |value| result.threads = value; } return result;}
fn parseKbField(line: []const u8, prefix: []const u8) ?u64 { if (!std.mem.startsWith(u8, line, prefix)) return null; const rest = std.mem.trim(u8, line[prefix.len..], " \t"); if (!std.mem.endsWith(u8, rest, "kB")) return null; const digits = std.mem.trim(u8, rest[0 .. rest.len - 2], " \t"); const kib = std.fmt.parseInt(u64, digits, 10) catch return null; return std.math.mul(u64, kib, 1024) catch null;}
fn parseIntegerField(line: []const u8, prefix: []const u8) ?u64 { if (!std.mem.startsWith(u8, line, prefix)) return null; return std.fmt.parseInt(u64, std.mem.trim(u8, line[prefix.len..], " \t"), 10) catch null;}
const DarwinTaskInfo = extern struct { virtual_size: u64, resident_size: u64, total_user: u64, total_system: u64, threads_user: u64, threads_system: u64, policy: i32, faults: i32, pageins: i32, cow_faults: i32, messages_sent: i32, messages_received: i32, syscalls_mach: i32, syscalls_unix: i32, context_switches: i32, thread_count: i32, running_thread_count: i32, priority: i32,};
const Darwin = struct { extern "c" fn proc_pidinfo(pid: std.posix.pid_t, flavor: c_int, arg: u64, buffer: *anyopaque, size: c_int) c_int;};
fn darwinTaskInfo() ?DarwinTaskInfo { if (@import("builtin").os.tag != .macos) return null; var info: DarwinTaskInfo = undefined; const written = Darwin.proc_pidinfo(std.posix.system.getpid(), 4, 0, &info, @sizeOf(DarwinTaskInfo)); if (written != @sizeOf(DarwinTaskInfo)) return null; return info;}
test "linux status parsing preserves procfs units" { const parsed = parseLinuxStatus("Name:\tstream\n" ++ "VmSize:\t 987654 kB\n" ++ "VmRSS:\t 12345 kB\n" ++ "Threads:\t17\n"); try std.testing.expectEqual(@as(?u64, 987654 * 1024), parsed.virtual_memory_bytes); try std.testing.expectEqual(@as(?u64, 12345 * 1024), parsed.resident_memory_bytes); try std.testing.expectEqual(@as(?u64, 17), parsed.threads);}
test "malformed procfs values remain unavailable" { const parsed = parseLinuxStatus("VmSize:\tnope kB\n" ++ "VmRSS:\t18446744073709551615 kB\n" ++ "Threads:\t-1\n"); try std.testing.expectEqual(@as(?u64, null), parsed.virtual_memory_bytes); try std.testing.expectEqual(@as(?u64, null), parsed.resident_memory_bytes); try std.testing.expectEqual(@as(?u64, null), parsed.threads);}
test "live process snapshot has real runtime measurements" { const snapshot = collect(std.testing.io); try std.testing.expect(snapshot.cpu_us != null); try std.testing.expect(snapshot.resident_memory_bytes != null); try std.testing.expect(snapshot.virtual_memory_bytes != null); try std.testing.expect(snapshot.threads != null and snapshot.threads.? > 0); try std.testing.expect(snapshot.open_fds != null and snapshot.open_fds.? > 0); try std.testing.expect(snapshot.max_fds != null and snapshot.max_fds.? >= snapshot.open_fds.?); try std.testing.expect(snapshot.minor_page_faults != null); try std.testing.expect(snapshot.major_page_faults != null); try std.testing.expect(snapshot.voluntary_context_switches != null); try std.testing.expect(snapshot.involuntary_context_switches != null);}
// All-arena glibc heap stats, parsed from malloc_info() XML — mallinfo()// covers the main arena only and its dashboard understated multi-GiB arena// fragmentation until the gauges below existed. Ported from zlay// (src/internal/broadcaster.zig, incident-2026-05-30-arena-bloat.md):// "balanced allocs + growing RSS = glibc arena fragmentation, not a leak."// heap_count counts <heap nr=...> sections: chained 64 MiB per-arena heaps,// the unit that actually pins RSS (MALLOC_ARENA_MAX caps arenas, not heaps).const MallocInfo = struct { system_bytes: u64, free_bytes: u64, heap_count: u32,};
const CFile = opaque {};const open_memstream = @extern(*const fn (ptr: *[*c]u8, size: *usize) callconv(.c) ?*CFile, .{ .name = "open_memstream" });const c_fclose = @extern(*const fn (*CFile) callconv(.c) c_int, .{ .name = "fclose" });const c_free = @extern(*const fn (?*anyopaque) callconv(.c) void, .{ .name = "free" });const malloc_info_fn = @extern(*const fn (options: c_int, stream: *CFile) callconv(.c) c_int, .{ .name = "malloc_info" });
fn readMallocInfo() ?MallocInfo { // comptime prunes the glibc-only externs from analysis on other hosts. return if (comptime @import("builtin").os.tag == .linux) readMallocInfoLinux() else null;}
fn readMallocInfoLinux() ?MallocInfo { var buf_ptr: [*c]u8 = null; var buf_size: usize = 0; const stream = open_memstream(&buf_ptr, &buf_size) orelse return null; _ = malloc_info_fn(0, stream); _ = c_fclose(stream); // flushes and populates buf_ptr/buf_size const p = buf_ptr orelse return null; defer c_free(p); const xml = p[0..buf_size]; return MallocInfo{ // top-level totals are the LAST occurrence (each <heap> emits its own) .system_bytes = lastSizeAttr(xml, "system type=\"current\"") orelse 0, .free_bytes = (lastSizeAttr(xml, "total type=\"fast\"") orelse 0) + (lastSizeAttr(xml, "total type=\"rest\"") orelse 0), .heap_count = countOccurrences(xml, "<heap nr="), };}
// find the last occurrence of `marker`, then parse the first `size="N"` after it.fn lastSizeAttr(xml: []const u8, marker: []const u8) ?u64 { const at = std.mem.lastIndexOf(u8, xml, marker) orelse return null; const rest = xml[at + marker.len ..]; const key = "size=\""; const kpos = std.mem.indexOf(u8, rest, key) orelse return null; const num = rest[kpos + key.len ..]; const end = std.mem.indexOfScalar(u8, num, '"') orelse return null; return std.fmt.parseInt(u64, num[0..end], 10) catch null;}
fn countOccurrences(haystack: []const u8, needle: []const u8) u32 { var n: u32 = 0; var i: usize = 0; while (std.mem.indexOfPos(u8, haystack, i, needle)) |pos| { n += 1; i = pos + needle.len; } return n;}
test "readMallocInfo parses all-arena totals" { // mirrors zlay's readMallocInfo test; only meaningful against glibc if (comptime @import("builtin").os.tag != .linux) return error.SkipZigTest; const a = std.heap.c_allocator; var kept: [64][]u8 = undefined; for (&kept) |*slot| slot.* = try a.alloc(u8, 4096); defer for (kept) |s| a.free(s); const info = readMallocInfo() orelse return error.MallocInfoUnavailable; try std.testing.expect(info.system_bytes > 0); try std.testing.expect(info.heap_count >= 1);}
test "lastSizeAttr / countOccurrences against sample XML" { // mirrors zlay's fixture: per-heap totals precede the top-level totals, // and the parser must take the LAST occurrence const xml = \\<malloc version="1"> \\<heap nr="0"> \\<total type="fast" count="1" size="100"/> \\<total type="rest" count="2" size="200"/> \\<system type="current" size="9999"/> \\</heap> \\<heap nr="1"> \\<total type="fast" count="3" size="111"/> \\<total type="rest" count="4" size="222"/> \\<system type="current" size="8888"/> \\</heap> \\<total type="fast" count="4" size="211"/> \\<total type="rest" count="6" size="422"/> \\<system type="current" size="18887"/> \\</malloc> ; try std.testing.expectEqual(@as(?u64, 18887), lastSizeAttr(xml, "system type=\"current\"")); try std.testing.expectEqual(@as(?u64, 211), lastSizeAttr(xml, "total type=\"fast\"")); try std.testing.expectEqual(@as(?u64, 422), lastSizeAttr(xml, "total type=\"rest\"")); try std.testing.expectEqual(@as(u32, 2), countOccurrences(xml, "<heap nr="));}