Add production fsession latency tracer

.github/workflows/test-libbpf.yml

@@ -20,6 +20,7 @@ jobs:
         submodules: 'recursive'
     - name: install deps
       run: |
+        sudo apt-get update
         sudo apt-get install -y --no-install-recommends \
         libelf1 libelf-dev zlib1g-dev \
         make git clang llvm pkg-config build-essential dwarves \
@@ -179,6 +180,10 @@ jobs:
       run: |
           make -C src/50-tcx
 
+    - name: test 52 fsession latency
+      run: |
+          make -C src/52-fsession-latency
+
     - name: test features bpf_token
       run: |
           make -C src/features/bpf_token

src/52-fsession-latency/.config

@@ -0,0 +1,10 @@
+level=Depth
+type=Tracing
+kernel_min=7.0
+kernel_min_basis=documented
+architectures=x86_64
+btf=required
+kernel_config=CONFIG_BPF=y,CONFIG_BPF_SYSCALL=y,CONFIG_BPF_JIT=y,CONFIG_BPF_EVENTS=y,CONFIG_DEBUG_INFO_BTF=y,CONFIG_DYNAMIC_FTRACE_WITH_DIRECT_CALLS=y
+hardware=none
+root=required
+test_status=ci-build

src/52-fsession-latency/.gitignore

@@ -0,0 +1,15 @@
+# Build artifacts
+.output/
+*.o
+*.skel.h
+
+# Generated binary
+fsession_latency
+
+# Python cache
+__pycache__/
+
+# Editor files
+*.swp
+*~
+.vscode/

src/52-fsession-latency/Makefile

@@ -0,0 +1,92 @@
+# SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)
+OUTPUT := .output
+CLANG ?= clang
+LIBBPF_SRC := $(abspath ../third_party/libbpf/src)
+BPFTOOL_SRC := $(abspath ../third_party/bpftool/src)
+LIBBPF_OBJ := $(abspath $(OUTPUT)/libbpf.a)
+BPFTOOL_OUTPUT ?= $(abspath $(OUTPUT)/bpftool)
+BPFTOOL ?= $(BPFTOOL_OUTPUT)/bootstrap/bpftool
+ARCH ?= $(shell uname -m | sed 's/x86_64/x86/' \
+			 | sed 's/arm.*/arm/' \
+			 | sed 's/aarch64/arm64/' \
+			 | sed 's/ppc64le/powerpc/' \
+			 | sed 's/mips.*/mips/' \
+			 | sed 's/riscv64/riscv/' \
+			 | sed 's/loongarch64/loongarch/')
+VMLINUX := ../third_party/vmlinux/$(ARCH)/vmlinux.h
+INCLUDES := -I$(OUTPUT) -I../third_party/libbpf/include/uapi -I$(dir $(VMLINUX)) -I.
+CFLAGS := -g -Wall -Wextra
+ALL_LDFLAGS := $(LDFLAGS) $(EXTRA_LDFLAGS)
+
+APPS = fsession_latency
+
+CLANG_BPF_SYS_INCLUDES ?= $(shell $(CLANG) -v -E - </dev/null 2>&1 \
+	| sed -n '/<...> search starts here:/,/End of search list./{ s| \(/.*\)|-idirafter \1|p }')
+
+ifeq ($(V),1)
+	Q =
+	msg =
+else
+	Q = @
+	msg = @printf '  %-8s %s%s\n' "$(1)" \
+		      "$(patsubst $(abspath $(OUTPUT))/%,%,$(2))" \
+		      "$(if $(3), $(3))";
+	MAKEFLAGS += --no-print-directory
+endif
+
+define allow-override
+  $(if $(or $(findstring environment,$(origin $(1))),\
+            $(findstring command line,$(origin $(1)))),,\
+    $(eval $(1) = $(2)))
+endef
+
+$(call allow-override,CC,$(CROSS_COMPILE)cc)
+$(call allow-override,LD,$(CROSS_COMPILE)ld)
+
+.PHONY: all clean test
+all: $(APPS)
+
+clean:
+	$(call msg,CLEAN)
+	$(Q)rm -rf $(OUTPUT) $(APPS)
+
+test: all
+	python3 tests/test_fsession_latency.py ./fsession_latency
+
+$(OUTPUT) $(OUTPUT)/libbpf $(BPFTOOL_OUTPUT):
+	$(call msg,MKDIR,$@)
+	$(Q)mkdir -p $@
+
+$(LIBBPF_OBJ): $(wildcard $(LIBBPF_SRC)/*.[ch] $(LIBBPF_SRC)/Makefile) | $(OUTPUT)/libbpf
+	$(call msg,LIB,$@)
+	$(Q)$(MAKE) -C $(LIBBPF_SRC) BUILD_STATIC_ONLY=1 \
+		    OBJDIR=$(dir $@)/libbpf DESTDIR=$(dir $@) \
+		    INCLUDEDIR= LIBDIR= UAPIDIR= install
+
+$(BPFTOOL): | $(BPFTOOL_OUTPUT)
+	$(call msg,BPFTOOL,$@)
+	$(Q)$(MAKE) ARCH= CROSS_COMPILE= OUTPUT=$(BPFTOOL_OUTPUT)/ -C $(BPFTOOL_SRC) bootstrap
+
+$(OUTPUT)/%.bpf.o: %.bpf.c $(LIBBPF_OBJ) $(wildcard %.h) $(VMLINUX) | $(OUTPUT) $(BPFTOOL)
+	$(call msg,BPF,$@)
+	$(Q)$(CLANG) -g -O2 -target bpf -D__TARGET_ARCH_$(ARCH) \
+		     $(INCLUDES) $(CLANG_BPF_SYS_INCLUDES) \
+		     -c $(filter %.c,$^) -o $(patsubst %.bpf.o,%.tmp.bpf.o,$@)
+	$(Q)$(BPFTOOL) gen object $@ $(patsubst %.bpf.o,%.tmp.bpf.o,$@)
+
+$(OUTPUT)/%.skel.h: $(OUTPUT)/%.bpf.o | $(OUTPUT) $(BPFTOOL)
+	$(call msg,GEN-SKEL,$@)
+	$(Q)$(BPFTOOL) gen skeleton $< > $@
+
+$(patsubst %,$(OUTPUT)/%.o,$(APPS)): %.o: %.skel.h
+
+$(OUTPUT)/%.o: %.c $(wildcard %.h) | $(OUTPUT)
+	$(call msg,CC,$@)
+	$(Q)$(CC) $(CFLAGS) $(INCLUDES) -c $(filter %.c,$^) -o $@
+
+$(APPS): %: $(OUTPUT)/%.o $(LIBBPF_OBJ) | $(OUTPUT)
+	$(call msg,BINARY,$@)
+	$(Q)$(CC) $(CFLAGS) $^ $(ALL_LDFLAGS) -lelf -lz -o $@
+
+.DELETE_ON_ERROR:
+.SECONDARY:

src/52-fsession-latency/fsession_latency.bpf.c

@@ -0,0 +1,74 @@
+// SPDX-License-Identifier: GPL-2.0
+#define bpf_session_is_return bpf_session_is_return_vmlinux_snapshot
+#define bpf_session_cookie bpf_session_cookie_vmlinux_snapshot
+#include "vmlinux.h"
+#undef bpf_session_is_return
+#undef bpf_session_cookie
+#include <bpf/bpf_helpers.h>
+#include <bpf/bpf_tracing.h>
+#include "fsession_latency.h"
+
+char LICENSE[] SEC("license") = "GPL";
+
+const volatile __u64 threshold_ns;
+const volatile __u32 target_tgid;
+
+struct latency_stats stats;
+
+struct {
+	__uint(type, BPF_MAP_TYPE_RINGBUF);
+	__uint(max_entries, 256 * 1024);
+} events SEC(".maps");
+
+/*
+ * The repository vmlinux.h snapshot predates the ctx argument on these
+ * kfunc prototypes. Rename those stale declarations while including the
+ * snapshot, then provide the Linux 7.0 signatures below.
+ */
+extern bool bpf_session_is_return(void *ctx) __ksym;
+extern __u64 *bpf_session_cookie(void *ctx) __ksym;
+
+SEC("fsession/vfs_read")
+int BPF_PROG(measure_vfs_read, struct file *file, char *buf, size_t count,
+	     loff_t *pos, ssize_t ret)
+{
+	__u64 pid_tgid = bpf_get_current_pid_tgid();
+	__u64 *started = bpf_session_cookie(ctx);
+	struct latency_event *event;
+	__u64 latency;
+
+	if (!bpf_session_is_return(ctx)) {
+		if (target_tgid && pid_tgid >> 32 != target_tgid) {
+			*started = 0;
+			return 0;
+		}
+		*started = bpf_ktime_get_ns();
+		return 0;
+	}
+
+	if (!*started)
+		return 0;
+
+	latency = bpf_ktime_get_ns() - *started;
+	__sync_fetch_and_add(&stats.calls, 1);
+	if (ret < 0)
+		__sync_fetch_and_add(&stats.errors, 1);
+	if (latency < threshold_ns)
+		return 0;
+
+	__sync_fetch_and_add(&stats.slow, 1);
+	event = bpf_ringbuf_reserve(&events, sizeof(*event), 0);
+	if (!event) {
+		__sync_fetch_and_add(&stats.dropped, 1);
+		return 0;
+	}
+
+	event->pid = (__u32)pid_tgid;
+	event->tgid = pid_tgid >> 32;
+	event->requested = count;
+	event->result = ret;
+	event->latency_ns = latency;
+	bpf_get_current_comm(event->comm, sizeof(event->comm));
+	bpf_ringbuf_submit(event, 0);
+	return 0;
+}

src/52-fsession-latency/fsession_latency.c

@@ -0,0 +1,237 @@
+// SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause)
+#include <errno.h>
+#include <getopt.h>
+#include <signal.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <bpf/libbpf.h>
+#include "fsession_latency.h"
+#include "fsession_latency.skel.h"
+
+static volatile sig_atomic_t exiting;
+
+static struct env {
+	unsigned long long threshold_us;
+	unsigned int duration;
+	unsigned int pid;
+	bool verbose;
+} env = {
+	.threshold_us = 1000,
+	.duration = 10,
+};
+
+static unsigned long long events_printed;
+
+static void handle_signal(int signal)
+{
+	(void)signal;
+	exiting = 1;
+}
+
+static int libbpf_print_fn(enum libbpf_print_level level, const char *format,
+			   va_list args)
+{
+	if (level == LIBBPF_DEBUG && !env.verbose)
+		return 0;
+	return vfprintf(stderr, format, args);
+}
+
+static void usage(const char *program)
+{
+	fprintf(stderr,
+		"Usage: %s [--threshold-us USEC] [--duration SEC] [--pid TGID] "
+		"[--verbose]\n\n"
+		"Report vfs_read calls at or above a latency threshold.\n\n"
+		"Options:\n"
+		"  -t, --threshold-us USEC  slow-read threshold (default: 1000)\n"
+		"  -d, --duration SEC       trace duration, 1-86400 (default: 10)\n"
+		"  -p, --pid TGID           trace one process ID (default: all)\n"
+		"  -v, --verbose            print libbpf diagnostics\n"
+		"  -h, --help               show this help\n",
+		program);
+}
+
+static int parse_u64(const char *value, unsigned long long maximum,
+		     unsigned long long *result)
+{
+	char *end = NULL;
+	unsigned long long parsed;
+
+	errno = 0;
+	parsed = strtoull(value, &end, 10);
+	if (errno || !end || *end || parsed > maximum)
+		return -EINVAL;
+	*result = parsed;
+	return 0;
+}
+
+static int parse_args(int argc, char **argv)
+{
+	static const struct option options[] = {
+		{ "threshold-us", required_argument, NULL, 't' },
+		{ "duration", required_argument, NULL, 'd' },
+		{ "pid", required_argument, NULL, 'p' },
+		{ "verbose", no_argument, NULL, 'v' },
+		{ "help", no_argument, NULL, 'h' },
+		{},
+	};
+	unsigned long long parsed;
+	int option;
+
+	while ((option = getopt_long(argc, argv, "t:d:p:vh", options, NULL)) != -1) {
+		switch (option) {
+		case 't':
+			if (parse_u64(optarg, UINT64_MAX / 1000, &env.threshold_us)) {
+				fprintf(stderr, "invalid threshold in microseconds: %s\n", optarg);
+				return -EINVAL;
+			}
+			break;
+		case 'd':
+			if (parse_u64(optarg, 86400, &parsed) || parsed == 0) {
+				fprintf(stderr, "invalid duration in seconds: %s\n", optarg);
+				return -EINVAL;
+			}
+			env.duration = parsed;
+			break;
+		case 'p':
+			if (parse_u64(optarg, UINT32_MAX, &parsed) || parsed == 0) {
+				fprintf(stderr, "invalid process ID: %s\n", optarg);
+				return -EINVAL;
+			}
+			env.pid = parsed;
+			break;
+		case 'v':
+			env.verbose = true;
+			break;
+		case 'h':
+			usage(argv[0]);
+			exit(0);
+		default:
+			return -EINVAL;
+		}
+	}
+
+	if (optind != argc)
+		return -EINVAL;
+	return 0;
+}
+
+static long long monotonic_milliseconds(void)
+{
+	struct timespec timestamp;
+
+	if (clock_gettime(CLOCK_MONOTONIC, &timestamp))
+		return -errno;
+	return timestamp.tv_sec * 1000LL + timestamp.tv_nsec / 1000000;
+}
+
+static int handle_event(void *context, void *data, size_t size)
+{
+	const struct latency_event *event = data;
+
+	(void)context;
+	if (size < sizeof(*event))
+		return 0;
+
+	printf("EVENT comm=%-16s tgid=%u pid=%u requested=%llu result=%lld "
+	       "latency_us=%llu\n",
+	       event->comm, event->tgid, event->pid, event->requested,
+	       event->result, event->latency_ns / 1000);
+	events_printed++;
+	return 0;
+}
+
+int main(int argc, char **argv)
+{
+	struct fsession_latency_bpf *skel = NULL;
+	struct ring_buffer *ring = NULL;
+	long long deadline, now;
+	int err;
+
+	err = parse_args(argc, argv);
+	if (err) {
+		usage(argv[0]);
+		return 1;
+	}
+
+	libbpf_set_print(libbpf_print_fn);
+	signal(SIGINT, handle_signal);
+	signal(SIGTERM, handle_signal);
+
+	skel = fsession_latency_bpf__open();
+	if (!skel) {
+		fprintf(stderr, "failed to open BPF skeleton\n");
+		return 1;
+	}
+
+	skel->rodata->threshold_ns = env.threshold_us * 1000;
+	skel->rodata->target_tgid = env.pid;
+
+	err = fsession_latency_bpf__load(skel);
+	if (err) {
+		fprintf(stderr,
+			"failed to load fsession program: %s\n"
+			"This tool requires Linux 7.0+, BTF, BPF JIT, and x86_64 "
+			"fsession support.\n",
+			strerror(-err));
+		goto cleanup;
+	}
+
+	err = fsession_latency_bpf__attach(skel);
+	if (err) {
+		fprintf(stderr, "failed to attach to vfs_read: %s\n", strerror(-err));
+		goto cleanup;
+	}
+
+	ring = ring_buffer__new(bpf_map__fd(skel->maps.events), handle_event, NULL, NULL);
+	err = libbpf_get_error(ring);
+	if (err) {
+		ring = NULL;
+		fprintf(stderr, "failed to create ring buffer: %s\n", strerror(-err));
+		goto cleanup;
+	}
+
+	printf("Tracing vfs_read for %u seconds; threshold=%llu us; pid=%s\n",
+	       env.duration, env.threshold_us, env.pid ? "selected" : "all");
+	fflush(stdout);
+
+	now = monotonic_milliseconds();
+	if (now < 0) {
+		err = now;
+		goto cleanup;
+	}
+	deadline = now + env.duration * 1000LL;
+
+	while (!exiting) {
+		now = monotonic_milliseconds();
+		if (now < 0) {
+			err = now;
+			break;
+		}
+		if (now >= deadline)
+			break;
+
+		err = ring_buffer__poll(ring, deadline - now > 100 ? 100 : deadline - now);
+		if (err == -EINTR) {
+			err = 0;
+			continue;
+		}
+		if (err < 0) {
+			fprintf(stderr, "ring buffer poll failed: %s\n", strerror(-err));
+			break;
+		}
+	}
+
+	printf("SUMMARY calls=%llu slow=%llu errors=%llu dropped=%llu events=%llu\n",
+	       skel->bss->stats.calls, skel->bss->stats.slow,
+	       skel->bss->stats.errors, skel->bss->stats.dropped, events_printed);
+
+cleanup:
+	ring_buffer__free(ring);
+	fsession_latency_bpf__destroy(skel);
+	return err != 0;
+}

src/52-fsession-latency/fsession_latency.h

@@ -0,0 +1,23 @@
+/* SPDX-License-Identifier: (LGPL-2.1 OR BSD-2-Clause) */
+#ifndef __FSESSION_LATENCY_H
+#define __FSESSION_LATENCY_H
+
+#define FSESSION_COMM_LEN 16
+
+struct latency_event {
+	unsigned int pid;
+	unsigned int tgid;
+	unsigned long long requested;
+	long long result;
+	unsigned long long latency_ns;
+	char comm[FSESSION_COMM_LEN];
+};
+
+struct latency_stats {
+	unsigned long long calls;
+	unsigned long long slow;
+	unsigned long long errors;
+	unsigned long long dropped;
+};
+
+#endif /* __FSESSION_LATENCY_H */

src/52-fsession-latency/tests/test_fsession_latency.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+"""Deterministic runtime checks for the fsession latency tool."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+import re
+import select
+import subprocess
+import sys
+import threading
+import time
+
+
+SUMMARY = re.compile(
+    r"SUMMARY calls=(\d+) slow=(\d+) errors=(\d+) dropped=(\d+) events=(\d+)"
+)
+
+
+def wait_until_attached(process: subprocess.Popen[str]) -> list[str]:
+    lines: list[str] = []
+    deadline = time.monotonic() + 8
+    assert process.stdout is not None
+
+    while time.monotonic() < deadline:
+        readable, _, _ = select.select([process.stdout], [], [], 0.25)
+        if readable:
+            line = process.stdout.readline()
+            if line:
+                lines.append(line)
+                if line.startswith("Tracing vfs_read"):
+                    return lines
+        if process.poll() is not None:
+            break
+
+    process.kill()
+    remainder, _ = process.communicate()
+    raise AssertionError("tool did not attach:\n" + "".join(lines) + remainder)
+
+
+def generate_reads() -> None:
+    path = pathlib.Path(__file__)
+    descriptor = os.open(path, os.O_RDONLY)
+    try:
+        for _ in range(64):
+            os.lseek(descriptor, 0, os.SEEK_SET)
+            os.read(descriptor, 256)
+    finally:
+        os.close(descriptor)
+
+    read_fd, write_fd = os.pipe()
+
+    def delayed_write() -> None:
+        time.sleep(0.05)
+        os.write(write_fd, b"x")
+        os.close(write_fd)
+
+    writer = threading.Thread(target=delayed_write)
+    writer.start()
+    try:
+        assert os.read(read_fd, 1) == b"x"
+    finally:
+        os.close(read_fd)
+        writer.join()
+
+
+def run_trace(binary: str, *arguments: str, workload: bool) -> tuple[dict[str, int], str]:
+    process = subprocess.Popen(
+        [binary, *arguments],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,
+    )
+    lines = wait_until_attached(process)
+    if workload:
+        generate_reads()
+    remainder, _ = process.communicate(timeout=8)
+    output = "".join(lines) + remainder
+    if process.returncode != 0:
+        raise AssertionError(f"tool failed with {process.returncode}:\n{output}")
+
+    match = SUMMARY.search(output)
+    if not match:
+        raise AssertionError("missing summary:\n" + output)
+    values = dict(zip(("calls", "slow", "errors", "dropped", "events"), map(int, match.groups())))
+    return values, output
+
+
+def main() -> int:
+    binary = sys.argv[1] if len(sys.argv) > 1 else "./fsession_latency"
+
+    invalid = subprocess.run(
+        [binary, "--duration", "invalid"],
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        check=False,
+    )
+    assert invalid.returncode != 0
+    assert "invalid duration" in invalid.stdout
+
+    pid = str(os.getpid())
+    missed, _ = run_trace(
+        binary,
+        "--duration", "1",
+        "--threshold-us", "0",
+        "--pid", str(2**32 - 1),
+        workload=True,
+    )
+    assert missed == {"calls": 0, "slow": 0, "errors": 0, "dropped": 0, "events": 0}, missed
+
+    below_threshold, _ = run_trace(
+        binary,
+        "--duration", "1",
+        "--threshold-us", "10000000",
+        "--pid", pid,
+        workload=True,
+    )
+    assert below_threshold["calls"] >= 64, below_threshold
+    assert below_threshold["slow"] == 0, below_threshold
+    assert below_threshold["events"] == 0, below_threshold
+
+    slow, slow_output = run_trace(
+        binary,
+        "--duration", "1",
+        "--threshold-us", "10000",
+        "--pid", pid,
+        workload=True,
+    )
+    assert slow["calls"] >= 65, slow
+    assert 1 <= slow["slow"] < slow["calls"], slow
+    assert slow["events"] > 0, slow
+    assert "EVENT " in slow_output
+
+    tracing_line = next(line for line in slow_output.splitlines() if line.startswith("Tracing "))
+    first_event = next(line for line in slow_output.splitlines() if line.startswith("EVENT "))
+    tool_summary = next(line for line in slow_output.splitlines() if line.startswith("SUMMARY "))
+    print(tracing_line)
+    print(first_event)
+    print(tool_summary)
+
+    print(
+        "TEST-SUMMARY miss_calls=0 "
+        f"high_threshold_calls={below_threshold['calls']} high_threshold_slow=0 "
+        f"threshold_calls={slow['calls']} threshold_slow={slow['slow']} "
+        f"events={slow['events']} dropped={slow['dropped']}"
+    )
+    print("PASS: PID filtering, threshold miss, and slow-read reporting behaved as expected")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())