Provide tests to exercise telemetry based packet capture. Signed-off-by: Stephen Hemminger <[email protected]> --- MAINTAINERS | 2 + app/test/meson.build | 1 + app/test/test_capture.c | 512 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 515 insertions(+) create mode 100644 app/test/test_capture.c
diff --git a/MAINTAINERS b/MAINTAINERS index fcd350ad94..5ce969d02f 100644 --- a/MAINTAINERS +++ b/MAINTAINERS @@ -1721,6 +1721,8 @@ Packet capture M: Reshma Pattan <[email protected]> M: Stephen Hemminger <[email protected]> F: lib/capture/ +F: usertools/dpdk-wireshark-extcap.py +F: doc/guides/tools/wireshark_extcap.rst F: lib/pdump/ F: doc/guides/prog_guide/pdump_lib.rst F: app/test/test_pdump.* diff --git a/app/test/meson.build b/app/test/meson.build index 51abeeb732..b73ae8ca07 100644 --- a/app/test/meson.build +++ b/app/test/meson.build @@ -37,6 +37,7 @@ source_file_deps = { 'test_bpf.c': ['bpf', 'net'], 'test_bpf_validate.c': ['bpf'], 'test_byteorder.c': [], + 'test_capture.c': ['net_null', 'net', 'ethdev', 'bus_vdev', 'capture'], 'test_cfgfile.c': ['cfgfile'], 'test_cksum.c': ['net'], 'test_cksum_perf.c': ['net'], diff --git a/app/test/test_capture.c b/app/test/test_capture.c new file mode 100644 index 0000000000..c37b2a961d --- /dev/null +++ b/app/test/test_capture.c @@ -0,0 +1,512 @@ +/* SPDX-License-Identifier: BSD-3-Clause + * Copyright(c) 2026 Stephen Hemminger + */ + +/* + * Functional test for the capture library. + * + * The capture library has no public C API: it is driven entirely through the + * telemetry socket. The output is a path ('out=') that the primary opens + * itself and writes pcapng into; the caller never receives packet data over + * the control socket. This test therefore behaves like an external capture + * tool. It: + * + * 1. builds a virtual ethdev backed by net_null. Rx synthesizes packets and + * Tx is a sink that frees whatever it is handed, which is all this test + * needs and avoids the per-queue ring bookkeeping of net_ring. The port + * is given NB_QUEUES queues so per-queue selection can be exercised; + * 2. creates a FIFO and opens its read end (so the primary's non-blocking + * O_WRONLY open finds a reader); + * 3. connects to this process's own telemetry socket; + * 4. runs three phases: + * - lifecycle: start an all-queue capture, check a pcapng stream + * appears, the capture is listed, stats account for traffic injected + * on every queue in both directions, and it tears down when the + * reader disconnects; + * - parameters: an out-of-range queue index is rejected, not started; + * - selection: a queue=CAPTURE_QUEUE capture installs callbacks on that + * queue only, so traffic on any other queue is not captured while + * traffic on that one is, in both directions. The selected queue is + * neither the first nor the last of NB_QUEUES, so an off-by-one in + * either install loop is caught. Both directions are steerable here: + * the test names the queue in rte_eth_rx_burst()/rte_eth_tx_burst() + * and net_null sources or sinks on whichever queue it is handed. + * + * The test is skipped (not failed) if telemetry is not enabled or the net_null + * driver is not available. + */ + +#include <ctype.h> +#include <errno.h> +#include <fcntl.h> +#include <inttypes.h> +#include <limits.h> +#include <signal.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> + +#include <sys/select.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/un.h> + +#include <rte_bus_vdev.h> +#include <rte_cycles.h> +#include <rte_eal.h> +#include <rte_ethdev.h> +#include <rte_mbuf.h> + +#include "test.h" + +#define TELEMETRY_VERSION "v2" +#define CAPTURE_START "/ethdev/capture/start" +#define CAPTURE_LIST "/ethdev/capture/list" +#define CAPTURE_STATS "/ethdev/capture/stats" + +#define NULL_VDEV_NAME "net_null_capture" +#define NB_QUEUES 4 +#define CAPTURE_QUEUE 2 /* not first or last of NB_QUEUES */ +#define RING_SIZE 256 +#define NB_MBUFS 1024 +#define MBUF_CACHE 32 +#define NB_PKTS 32 +#define PKT_LEN 64 +#define REPLY_LEN 16384 + +/* pcapng Section Header Block type, byte-order independent on disk. */ +static const uint8_t pcapng_shb_magic[4] = { 0x0a, 0x0d, 0x0d, 0x0a }; + +static struct rte_mempool *test_mp; +static uint16_t test_port = RTE_MAX_ETHPORTS; +static char fifo_path[PATH_MAX]; + +/* --- telemetry client helpers ------------------------------------------ */ + +/* Connect to this process's telemetry socket; -1 (and skip) if unavailable. */ +static int +tel_connect(void) +{ + struct sockaddr_un addr = { .sun_family = AF_UNIX }; + char buf[REPLY_LEN]; + int s; + + snprintf(addr.sun_path, sizeof(addr.sun_path), "%s/dpdk_telemetry.%s", + rte_eal_get_runtime_dir(), TELEMETRY_VERSION); + + s = socket(AF_UNIX, SOCK_SEQPACKET, 0); + if (s < 0) + return -1; + + if (connect(s, (struct sockaddr *)&addr, sizeof(addr)) < 0) { + close(s); + return -1; + } + + /* Server greets with an info message; consume it. */ + if (recv(s, buf, sizeof(buf), 0) <= 0) { + close(s); + return -1; + } + return s; +} + +/* Send a command and read the reply. */ +static int +tel_cmd(int s, const char *cmd, char *reply, size_t reply_sz) +{ + ssize_t n; + + if (send(s, cmd, strlen(cmd), 0) < 0) + return -1; + n = recv(s, reply, reply_sz - 1, 0); + if (n < 0) + return -1; + reply[n] = '\0'; + return 0; +} + +/* Minimal JSON scanning: find "key" and read the unsigned number after it. */ +static int +json_uint(const char *s, const char *key, uint64_t *out) +{ + const char *p = strstr(s, key); + + if (p == NULL) + return -1; + for (p += strlen(key); *p != '\0' && !isdigit((unsigned char)*p); p++) + ; + if (*p == '\0') + return -1; + *out = strtoull(p, NULL, 10); + return 0; +} + +/* Read the first element of the array in a list reply; -1 if empty/absent. */ +static int +json_first_array_uint(const char *s, uint64_t *out) +{ + const char *p = strchr(s, '['); + + if (p == NULL) + return -1; + for (p++; *p == ' '; p++) + ; + if (*p == ']' || !isdigit((unsigned char)*p)) + return -1; + *out = strtoull(p, NULL, 10); + return 0; +} + +/* Query the accepted counter for a capture id via telemetry. */ +static int +capture_accepted(int sock, uint64_t id, uint64_t *accepted) +{ + char cmd[64], reply[REPLY_LEN]; + + snprintf(cmd, sizeof(cmd), "%s,%" PRIu64, CAPTURE_STATS, id); + if (tel_cmd(sock, cmd, reply, sizeof(reply)) < 0) + return -1; + return json_uint(reply, "\"accepted\"", accepted); +} + +/* + * Poll the capture list until it is empty (the capture has torn down). + * Returns 0 once empty, -1 on timeout (~2s) or telemetry failure. + */ +static int +wait_list_empty(int sock) +{ + char reply[REPLY_LEN]; + uint64_t id; + + for (int i = 0; i < 200; i++) { + if (tel_cmd(sock, CAPTURE_LIST, reply, sizeof(reply)) < 0) + return -1; + if (json_first_array_uint(reply, &id) < 0) + return 0; + rte_delay_ms(10); + } + return -1; +} + +/* --- packet injection --------------------------------------------------- */ + +/* + * Pull count packets from a queue. net_null synthesizes them and pulling runs + * the capture Rx callback on each; the mbufs are ours to free. + */ +static int +inject_rx(uint16_t queue, unsigned int count) +{ + struct rte_mbuf *bufs[NB_PKTS]; + uint16_t got; + + if (count > NB_PKTS) + count = NB_PKTS; + + got = rte_eth_rx_burst(test_port, queue, bufs, count); + rte_pktmbuf_free_bulk(bufs, got); + return got == count ? 0 : -1; /* net_null fills the whole request */ +} + +/* + * Transmit count packets on a queue. The capture Tx callback runs first (it + * only copies) and net_null then frees the originals, so nothing is reclaimed + * here. + */ +static int +inject_tx(uint16_t queue, unsigned int count) +{ + struct rte_mbuf *bufs[NB_PKTS]; + + if (count > NB_PKTS) + count = NB_PKTS; + + for (unsigned int i = 0; i < count; i++) { + struct rte_mbuf *m = rte_pktmbuf_alloc(test_mp); + + if (m == NULL) { + rte_pktmbuf_free_bulk(bufs, i); + return -1; + } + m->pkt_len = m->data_len = PKT_LEN; + memset(rte_pktmbuf_mtod(m, void *), 0, PKT_LEN); + bufs[i] = m; + } + + /* net_null accepts the whole burst. */ + return rte_eth_tx_burst(test_port, queue, bufs, count) == count ? 0 : -1; +} + +/* + * Inject on every queue except skip; pass NB_QUEUES to cover them all. + */ +static int +inject_rx_except(uint16_t skip, unsigned int count) +{ + for (uint16_t q = 0; q < NB_QUEUES; q++) { + if (q != skip && inject_rx(q, count) < 0) + return -1; + } + return 0; +} + +static int +inject_tx_except(uint16_t skip, unsigned int count) +{ + for (uint16_t q = 0; q < NB_QUEUES; q++) { + if (q != skip && inject_tx(q, count) < 0) + return -1; + } + return 0; +} + +/* --- fixture ------------------------------------------------------------ */ + +static int +build_port(void) +{ + struct rte_eth_conf conf = { 0 }; + + test_mp = rte_pktmbuf_pool_create("capture_test_mp", NB_MBUFS, MBUF_CACHE, + 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id()); + if (test_mp == NULL) + return -1; + + if (rte_vdev_init(NULL_VDEV_NAME, NULL) < 0) + return -1; + if (rte_eth_dev_get_port_by_name(NULL_VDEV_NAME, &test_port) != 0) + return -1; + + if (rte_eth_dev_configure(test_port, NB_QUEUES, NB_QUEUES, &conf) < 0) + return -1; + for (uint16_t q = 0; q < NB_QUEUES; q++) { + if (rte_eth_rx_queue_setup(test_port, q, RING_SIZE, rte_socket_id(), + NULL, test_mp) < 0) + return -1; + if (rte_eth_tx_queue_setup(test_port, q, RING_SIZE, rte_socket_id(), + NULL) < 0) + return -1; + } + + return rte_eth_dev_start(test_port); +} + +static void +teardown_port(void) +{ + if (test_port != RTE_MAX_ETHPORTS) { + rte_eth_dev_stop(test_port); + /* + * Uninit drives the driver's remove, which closes and releases + * the ethdev port and removes the vdev in one path. No separate + * rte_eth_dev_close() is needed and null's close is idempotent, + * so this is safe even if a tree's close already released it. + */ + rte_vdev_uninit(NULL_VDEV_NAME); + test_port = RTE_MAX_ETHPORTS; + } + rte_mempool_free(test_mp); + test_mp = NULL; +} + +/* Create the capture FIFO in the runtime dir; -1 on failure. */ +static int +make_fifo(void) +{ + snprintf(fifo_path, sizeof(fifo_path), "%s/capture_test.%d", + rte_eal_get_runtime_dir(), (int)getpid()); + unlink(fifo_path); /* clear any stale node */ + if (mkfifo(fifo_path, 0600) < 0) + return -1; + return 0; +} + +/* --- the test ----------------------------------------------------------- */ + +static int +test_capture(void) +{ + char cmd[PATH_MAX + 64], reply[REPLY_LEN], pcapng[REPLY_LEN]; + int sock = -1, rd = -1; + int ret = TEST_FAILED; + uint64_t id, accepted; + struct timeval tv; + fd_set rfds; + ssize_t n; + + fifo_path[0] = '\0'; + + /* + * The library writes to the FIFO; a closed reader must give the writer + * EPIPE, not a fatal SIGPIPE. The library masks SIGPIPE on its drain + * thread, but ignore it here too so the test process is safe regardless. + */ + signal(SIGPIPE, SIG_IGN); + + sock = tel_connect(); + if (sock < 0) { + printf("telemetry socket not available, skipping\n"); + return TEST_SKIPPED; + } + + if (build_port() < 0) { + printf("could not build net_null test port, skipping\n"); + ret = TEST_SKIPPED; + goto out; + } + + if (make_fifo() < 0) + goto out; + + /* + * Phase 1: all-queue lifecycle. + * + * Open the read end before starting: the primary opens the FIFO + * O_WRONLY|O_NONBLOCK and would get ENXIO with no reader present. + * O_RDONLY|O_NONBLOCK returns immediately even with no writer yet. + */ + rd = open(fifo_path, O_RDONLY | O_NONBLOCK); + if (rd < 0) + goto out; + + snprintf(cmd, sizeof(cmd), "%s,%u,out=%s", CAPTURE_START, test_port, fifo_path); + TEST_ASSERT_SUCCESS(tel_cmd(sock, cmd, reply, sizeof(reply)), + "capture start command failed"); + TEST_ASSERT(strstr(reply, "error") == NULL, + "capture start returned an error: %s", reply); + + /* + * Inject traffic on every queue: with no queue= the callbacks must be + * installed on all of them. Rx callbacks run synchronously inside + * rx_burst, so the accepted counter is up to date as soon as this + * returns. + */ + TEST_ASSERT_SUCCESS(inject_rx_except(NB_QUEUES, NB_PKTS), + "rx injection failed"); + + /* A pcapng stream (at least the section header) must appear. */ + FD_ZERO(&rfds); + FD_SET(rd, &rfds); + tv = (struct timeval){ .tv_sec = 2 }; + TEST_ASSERT(select(rd + 1, &rfds, NULL, NULL, &tv) > 0, + "no pcapng output within timeout"); + n = read(rd, pcapng, sizeof(pcapng)); + TEST_ASSERT(n >= 4, "short pcapng read (%zd)", n); + TEST_ASSERT(memcmp(pcapng, pcapng_shb_magic, sizeof(pcapng_shb_magic)) == 0, + "output does not start with a pcapng section header block"); + + /* The capture must show up in the list. */ + TEST_ASSERT_SUCCESS(tel_cmd(sock, CAPTURE_LIST, reply, sizeof(reply)), + "capture list command failed"); + TEST_ASSERT_SUCCESS(json_first_array_uint(reply, &id), + "no capture id in list reply: %s", reply); + + /* Stats must report exactly the packets we injected, on every queue. */ + TEST_ASSERT_SUCCESS(capture_accepted(sock, id, &accepted), + "capture stats query failed"); + TEST_ASSERT_EQUAL(accepted, (uint64_t)(NB_QUEUES * NB_PKTS), + "accepted %" PRIu64 " != %d", accepted, NB_QUEUES * NB_PKTS); + + /* Same for the Tx side of every queue. */ + TEST_ASSERT_SUCCESS(inject_tx_except(NB_QUEUES, NB_PKTS), + "tx injection failed"); + TEST_ASSERT_SUCCESS(capture_accepted(sock, id, &accepted), + "capture stats query failed"); + TEST_ASSERT_EQUAL(accepted, (uint64_t)(2 * NB_QUEUES * NB_PKTS), + "accepted %" PRIu64 " != %d", accepted, 2 * NB_QUEUES * NB_PKTS); + + /* + * Close the reader: the capture should detect the hangup and tear down. + * The drain thread's idle poll notices POLLERR on the write end on its + * own; the extra injection just shortens the wait if it was mid-drain. + */ + close(rd); + rd = -1; + inject_rx(0, NB_PKTS); /* any queue; all are captured here */ + TEST_ASSERT_SUCCESS(wait_list_empty(sock), + "capture did not tear down after reader closed"); + + /* + * Phase 2: parameter validation. An out-of-range queue index must be + * rejected. The range check runs before the output is opened, so no + * reader is needed and no capture should be left behind. + */ + snprintf(cmd, sizeof(cmd), "%s,%u,out=%s,queue=%u", + CAPTURE_START, test_port, fifo_path, NB_QUEUES); + TEST_ASSERT_SUCCESS(tel_cmd(sock, cmd, reply, sizeof(reply)), + "capture start command failed"); + TEST_ASSERT(strstr(reply, "error") != NULL, + "out-of-range queue=%u was not rejected: %s", NB_QUEUES, reply); + TEST_ASSERT_SUCCESS(wait_list_empty(sock), + "rejected start left a capture behind"); + + /* + * Phase 3: single-queue selection. With queue=CAPTURE_QUEUE the + * callbacks are installed on that queue only: traffic on any other + * queue must not be captured, traffic on that one must be, in both + * directions. The counters are bumped synchronously inside + * rx_burst/tx_burst, so every check here is race-free. + */ + rd = open(fifo_path, O_RDONLY | O_NONBLOCK); + TEST_ASSERT(rd >= 0, "reopen fifo failed: %s", strerror(errno)); + + snprintf(cmd, sizeof(cmd), "%s,%u,out=%s,queue=%u", + CAPTURE_START, test_port, fifo_path, CAPTURE_QUEUE); + TEST_ASSERT_SUCCESS(tel_cmd(sock, cmd, reply, sizeof(reply)), + "queue=%u capture start failed", CAPTURE_QUEUE); + TEST_ASSERT(strstr(reply, "error") == NULL, + "queue=%u capture start returned an error: %s", + CAPTURE_QUEUE, reply); + TEST_ASSERT_SUCCESS(json_uint(reply, "\"id\"", &id), + "no id in start reply: %s", reply); + + /* Unselected queues: callbacks never installed, nothing captured. */ + TEST_ASSERT_SUCCESS(inject_rx_except(CAPTURE_QUEUE, NB_PKTS), + "rx inject on unselected queues failed"); + TEST_ASSERT_SUCCESS(inject_tx_except(CAPTURE_QUEUE, NB_PKTS), + "tx inject on unselected queues failed"); + TEST_ASSERT_SUCCESS(capture_accepted(sock, id, &accepted), + "capture stats query failed"); + TEST_ASSERT_EQUAL(accepted, (uint64_t)0, + "other queues captured under queue=%u (accepted %" PRIu64 ")", + CAPTURE_QUEUE, accepted); + + /* Selected queue: captured, on Rx and on Tx. */ + TEST_ASSERT_SUCCESS(inject_rx(CAPTURE_QUEUE, NB_PKTS), + "rx inject on queue %u failed", CAPTURE_QUEUE); + TEST_ASSERT_SUCCESS(capture_accepted(sock, id, &accepted), + "capture stats query failed"); + TEST_ASSERT_EQUAL(accepted, (uint64_t)NB_PKTS, + "rx queue %u not captured (accepted %" PRIu64 ")", + CAPTURE_QUEUE, accepted); + + TEST_ASSERT_SUCCESS(inject_tx(CAPTURE_QUEUE, NB_PKTS), + "tx inject on queue %u failed", CAPTURE_QUEUE); + TEST_ASSERT_SUCCESS(capture_accepted(sock, id, &accepted), + "capture stats query failed"); + TEST_ASSERT_EQUAL(accepted, (uint64_t)(2 * NB_PKTS), + "tx queue %u not captured (accepted %" PRIu64 ")", + CAPTURE_QUEUE, accepted); + + close(rd); + rd = -1; + TEST_ASSERT_SUCCESS(wait_list_empty(sock), + "queue=%u capture did not tear down", CAPTURE_QUEUE); + + ret = TEST_SUCCESS; +out: + if (rd >= 0) + close(rd); + if (fifo_path[0] != '\0') + unlink(fifo_path); + if (sock >= 0) + close(sock); + teardown_port(); + return ret; +} + +REGISTER_FAST_TEST(capture_autotest, NOHUGE_OK, ASAN_OK, test_capture); -- 2.53.0
