#!/usr/bin/env python # Copyright (c) PLUMgrid, Inc. # Licensed under the Apache License, Version 2.0 (the "License") from bcc import BPF import os import ctypes as ct import random import time import subprocess from unittest import main, TestCase, skipUnless from utils import kernel_version_ge class TestRingbuf(TestCase): @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8") def test_ringbuf_output(self): self.counter = 0 class Data(ct.Structure): _fields_ = [("ts", ct.c_ulonglong)] def cb(ctx, data, size): self.assertEqual(size, ct.sizeof(Data)) event = ct.cast(data, ct.POINTER(Data)).contents self.counter += 1 text = """ BPF_RINGBUF_OUTPUT(events, 8); struct data_t { u64 ts; }; int do_sys_nanosleep(void *ctx) { struct data_t data = {bpf_ktime_get_ns()}; events.ringbuf_output(&data, sizeof(data), 0); return 0; } """ b = BPF(text=text) b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"), fn_name="do_sys_nanosleep") b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"), fn_name="do_sys_nanosleep") b["events"].open_ring_buffer(cb) subprocess.call(['sleep', '0.1']) b.ring_buffer_poll() self.assertGreater(self.counter, 0) b.cleanup() @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8") def test_ringbuf_consume(self): self.counter = 0 class Data(ct.Structure): _fields_ = [("ts", ct.c_ulonglong)] def cb(ctx, data, size): self.assertEqual(size, ct.sizeof(Data)) event = ct.cast(data, ct.POINTER(Data)).contents self.counter += 1 text = """ BPF_RINGBUF_OUTPUT(events, 8); struct data_t { u64 ts; }; int do_sys_nanosleep(void *ctx) { struct data_t data = {bpf_ktime_get_ns()}; events.ringbuf_output(&data, sizeof(data), 0); return 0; } """ b = BPF(text=text) b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"), fn_name="do_sys_nanosleep") b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"), fn_name="do_sys_nanosleep") b["events"].open_ring_buffer(cb) subprocess.call(['sleep', '0.1']) b.ring_buffer_consume() self.assertGreater(self.counter, 0) b.cleanup() @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8") def test_ringbuf_submit(self): self.counter = 0 class Data(ct.Structure): _fields_ = [("ts", ct.c_ulonglong)] def cb(ctx, data, size): self.assertEqual(size, ct.sizeof(Data)) event = ct.cast(data, ct.POINTER(Data)).contents self.counter += 1 text = """ BPF_RINGBUF_OUTPUT(events, 8); struct data_t { u64 ts; }; int do_sys_nanosleep(void *ctx) { struct data_t *data = events.ringbuf_reserve(sizeof(struct data_t)); if (!data) return 1; data->ts = bpf_ktime_get_ns(); events.ringbuf_submit(data, 0); return 0; } """ b = BPF(text=text) b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"), fn_name="do_sys_nanosleep") b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"), fn_name="do_sys_nanosleep") b["events"].open_ring_buffer(cb) subprocess.call(['sleep', '0.1']) b.ring_buffer_poll() self.assertGreater(self.counter, 0) b.cleanup() @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8") def test_ringbuf_discard(self): self.counter = 0 class Data(ct.Structure): _fields_ = [("ts", ct.c_ulonglong)] def cb(ctx, data, size): self.assertEqual(size, ct.sizeof(Data)) event = ct.cast(data, ct.POINTER(Data)).contents self.counter += 1 text = """ BPF_RINGBUF_OUTPUT(events, 8); struct data_t { u64 ts; }; int do_sys_nanosleep(void *ctx) { struct data_t *data = events.ringbuf_reserve(sizeof(struct data_t)); if (!data) return 1; data->ts = bpf_ktime_get_ns(); events.ringbuf_discard(data, 0); return 0; } """ b = BPF(text=text) b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"), fn_name="do_sys_nanosleep") b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"), fn_name="do_sys_nanosleep") b["events"].open_ring_buffer(cb) subprocess.call(['sleep', '0.1']) b.ring_buffer_poll() self.assertEqual(self.counter, 0) b.cleanup() if __name__ == "__main__": main()