From e84c7a2b8c1c513af124f22bb0703c75eff8301b Mon Sep 17 00:00:00 2001 From: "timmy.yuan" Date: Fri, 9 Oct 2026 10:05:51 +0800 Subject: [PATCH 1/2] fix(cpworker): count every output packet once, with its sliced length MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The output counters had no definition, and each output counted bytes differently: gre/vxlan added their encapsulation header to fwd, error and ratelimit bytes but not to direction drops, zmq counted whole batches including batch and record headers, a partial send split one packet across fwd_bytes and error_drop_bytes, and a VXLAN-split packet counted once per fragment. Each packet handed to an output is now counted once, in exactly one of fwd, direction_drop, ratelimit_drop or error_drop, with its caplen after slice and no encapsulation. A partial send, a failed split fragment or a failed fragment build counts the whole packet as error_drop. zmq batches track the frame bytes of their captured packets. rate_limit_mbps charges the same length, so fwd_bytes stays within the limit. The definition is written into REFACTOR-CPCTL.md §6.2, and the zmq wire format and usage docs refer to it. --- cpworker/src/output_gre.c | 13 +- cpworker/src/output_vxlan.c | 76 ++++----- cpworker/src/output_zmq.c | 17 +- cpworker/src/output_zmq.h | 4 +- cpworker/tests/unit/CMakeLists.txt | 1 + cpworker/tests/unit/output_vxlan.c | 240 +++++++++++++++++++++++++++++ cpworker/tests/unit/output_zmq.c | 95 +++++++++++- docs/REFACTOR-CPCTL.md | 33 ++++ docs/USAGE-CPWORKER-zh-Hans.md | 2 +- docs/USAGE-CPWORKER.md | 2 +- docs/ZMQ-WIRE-FORMAT-zh-Hans.md | 4 +- docs/ZMQ-WIRE-FORMAT.md | 6 +- 12 files changed, 428 insertions(+), 65 deletions(-) create mode 100644 cpworker/tests/unit/output_vxlan.c diff --git a/cpworker/src/output_gre.c b/cpworker/src/output_gre.c index e1cb4e4f..215d5ec5 100644 --- a/cpworker/src/output_gre.c +++ b/cpworker/src/output_gre.c @@ -59,9 +59,9 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const if (output->rate_limit_mbps > 0) { - if (!token_bucket_consume(&output->throttle, GRE_HEADER_LEN + length, header->ts)) + if (!token_bucket_consume(&output->throttle, length, header->ts)) { - bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, GRE_HEADER_LEN + length); + bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, length); packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1); return -1; } @@ -112,7 +112,7 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const output->error_info.nb_other_send_error_drops++; } - bytes_stats_add(&output->base.stats->error_drop_bytes, GRE_HEADER_LEN + length); + bytes_stats_add(&output->base.stats->error_drop_bytes, length); packets_stats_add(&output->base.stats->error_drop_packets, 1); return -1; } @@ -121,13 +121,12 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const { output->error_info.nb_partial_sends++; - bytes_stats_add(&output->base.stats->error_drop_bytes, GRE_HEADER_LEN + length - send_bytes); - bytes_stats_add(&output->base.stats->fwd_bytes, send_bytes); - packets_stats_add(&output->base.stats->fwd_packets, 1); + bytes_stats_add(&output->base.stats->error_drop_bytes, length); + packets_stats_add(&output->base.stats->error_drop_packets, 1); return -1; } - bytes_stats_add(&output->base.stats->fwd_bytes, GRE_HEADER_LEN + length); + bytes_stats_add(&output->base.stats->fwd_bytes, length); packets_stats_add(&output->base.stats->fwd_packets, 1); return 0; } while (true); diff --git a/cpworker/src/output_vxlan.c b/cpworker/src/output_vxlan.c index f55603fc..f24c66e6 100644 --- a/cpworker/src/output_vxlan.c +++ b/cpworker/src/output_vxlan.c @@ -93,6 +93,7 @@ static void flush_error_info(vxlan_output_t *output) } } +// Sends one VXLAN datagram. Returns 0 when it was sent in full, -1 otherwise. The caller does the counting. static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const uint8_t *pkt_data, size_t length, int direct) { @@ -159,28 +160,48 @@ static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const output->error_info.nb_other_send_error_drops++; } - - bytes_stats_add(&output->base.stats->error_drop_bytes, VXLAN_HEADER_LEN + length); - packets_stats_add(&output->base.stats->error_drop_packets, 1); return -1; } if (send_bytes < VXLAN_HEADER_LEN + length) { output->error_info.nb_partial_sends++; - - bytes_stats_add(&output->base.stats->error_drop_bytes, VXLAN_HEADER_LEN + length - send_bytes); - bytes_stats_add(&output->base.stats->fwd_bytes, send_bytes); - packets_stats_add(&output->base.stats->fwd_packets, 1); return -1; } - bytes_stats_add(&output->base.stats->fwd_bytes, VXLAN_HEADER_LEN + length); - packets_stats_add(&output->base.stats->fwd_packets, 1); return 0; } while (true); } +// Sends the packet, split into fragments when configured. Returns 0 only when every datagram was sent in full. +static int send_whole_or_split(vxlan_output_t *output, const struct pcap_pkthdr *header, const uint8_t *pkt_data, + int32_t caplen, size_t length, int direct) +{ + // Fast path + if (output->split.max_payload_size <= 0 || length <= (size_t)output->split.max_payload_size) + return do_send_packet(output, header->ts, pkt_data, length, direct); + + packet_parse_result_t parse_result; + if (!parse_packet(pkt_data, caplen, &parse_result)) + return do_send_packet(output, header->ts, pkt_data, length, direct); + + int fragment_count = calculate_fragment_count(&parse_result, output->split.max_payload_size); + if (fragment_count == 1) + return do_send_packet(output, header->ts, pkt_data, length, direct); + + for (int i = 0; i < fragment_count; i++) + { + int frag_len = build_fragment(&parse_result, pkt_data, i, output->split.max_payload_size, + output->split.recalculate_checksum, output->split.fragment_buf); + if (frag_len <= 0) + return -1; + + if (do_send_packet(output, header->ts, output->split.fragment_buf, frag_len, direct) != 0) + return -1; + } + return 0; +} + int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const uint8_t *pkt_data, int direct) { vxlan_output_t *output = (vxlan_output_t *)self; @@ -202,9 +223,9 @@ int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, con if (output->rate_limit_mbps > 0) { - if (!token_bucket_consume(&output->throttle, VXLAN_HEADER_LEN + length, header->ts)) + if (!token_bucket_consume(&output->throttle, length, header->ts)) { - bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, VXLAN_HEADER_LEN + length); + bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, length); packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1); return -1; } @@ -219,35 +240,16 @@ int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, con output->error_info.first_pktsec = header->ts.tv_sec; } - // Fast path - if (output->split.max_payload_size <= 0 || length <= (size_t)output->split.max_payload_size) + // A packet counts once, with its sliced length, whether or not it was split + if (send_whole_or_split(output, header, pkt_data, caplen, length, direct) != 0) { - return do_send_packet(output, header->ts, pkt_data, length, direct); - } - - packet_parse_result_t parse_result; - if (!parse_packet(pkt_data, caplen, &parse_result)) - { - return do_send_packet(output, header->ts, pkt_data, length, direct); - } - - int fragment_count = calculate_fragment_count(&parse_result, output->split.max_payload_size); - if (fragment_count == 1) - { - return do_send_packet(output, header->ts, pkt_data, length, direct); + bytes_stats_add(&output->base.stats->error_drop_bytes, length); + packets_stats_add(&output->base.stats->error_drop_packets, 1); + return -1; } - for (int i = 0; i < fragment_count; i++) - { - int frag_len = build_fragment(&parse_result, pkt_data, i, output->split.max_payload_size, - output->split.recalculate_checksum, output->split.fragment_buf); - if (frag_len <= 0) - return 0; - - int ret = do_send_packet(output, header->ts, output->split.fragment_buf, frag_len, direct); - if (ret != 0) - return ret; - } + bytes_stats_add(&output->base.stats->fwd_bytes, length); + packets_stats_add(&output->base.stats->fwd_packets, 1); return 0; } diff --git a/cpworker/src/output_zmq.c b/cpworker/src/output_zmq.c index 4beda450..59d745a1 100644 --- a/cpworker/src/output_zmq.c +++ b/cpworker/src/output_zmq.c @@ -113,10 +113,9 @@ int zmq_flush_packet(zmq_output_t *output) output->error_info.first_pktsec = pkts_buf->first_pktsec; } - // Only captured packets are counted as forwarded or dropped. A batch that carries - // nothing but heartbeats contributes no bytes, not even its batch header. + // Only captured packets are counted as forwarded or dropped, each with its counted length uint16_t data_num = send_num - pkts_buf->heartbeat_num; - uint32_t data_bytes = data_num > 0 ? pkts_buf->batch_bufpos - pkts_buf->heartbeat_bytes : 0; + uint32_t data_bytes = pkts_buf->data_bytes; int rc = zmq_send(output->pusher, &(pkts_buf->buf[0]), pkts_buf->batch_bufpos, ZMQ_DONTWAIT); if (rc != -1) @@ -141,7 +140,7 @@ int zmq_flush_packet(zmq_output_t *output) pkts_buf->batch_bufpos = sizeof(zmq_pkt_batch_hdr_t); pkts_buf->batch_hdr.pkts_num = 0; pkts_buf->heartbeat_num = 0; - pkts_buf->heartbeat_bytes = 0; + pkts_buf->data_bytes = 0; return 0; } @@ -188,7 +187,7 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const if (l2_len == 0) { output->error_info.nb_too_small_packets++; - bytes_stats_add(&output->base.stats->error_drop_bytes, header->caplen); + bytes_stats_add(&output->base.stats->error_drop_bytes, caplen); packets_stats_add(&output->base.stats->error_drop_packets, 1); zmq_flush_if_stale(output, header->ts.tv_sec); @@ -204,16 +203,16 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const if (direct == PKT_DIR_UNKNOWN) { - bytes_stats_add(&output->base.stats->direction_drop_bytes, length); + bytes_stats_add(&output->base.stats->direction_drop_bytes, caplen); packets_stats_add(&output->base.stats->direction_drop_packets, 1); zmq_flush_if_stale(output, header->ts.tv_sec); return -1; } - if (output->rate_limit_mbps > 0 && !token_bucket_consume(&output->throttle, length, header->ts)) + if (output->rate_limit_mbps > 0 && !token_bucket_consume(&output->throttle, caplen, header->ts)) { - bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, length); + bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, caplen); packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1); zmq_flush_if_stale(output, header->ts.tv_sec); @@ -294,6 +293,7 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const pkts_buf->batch_bufpos = buff_pos; pkts_buf->batch_hdr.pkts_num++; + pkts_buf->data_bytes += caplen; output->last_pkt_tv.tv_sec = header->ts.tv_sec; output->last_pkt_tv.tv_usec = header->ts.tv_usec; @@ -344,7 +344,6 @@ static void zmq_send_heartbeat_packet(zmq_output_t *output, struct timeval *tv) memcpy(&pkts_buf->buf[pos + offsetof(struct ether_header, ether_type)], &hb_etype, sizeof(hb_etype)); pos += sizeof(eth); - pkts_buf->heartbeat_bytes += pos - pkts_buf->batch_bufpos; pkts_buf->heartbeat_num++; pkts_buf->batch_bufpos = pos; pkts_buf->batch_hdr.pkts_num++; diff --git a/cpworker/src/output_zmq.h b/cpworker/src/output_zmq.h index 68fd7b5f..1af3b9e9 100644 --- a/cpworker/src/output_zmq.h +++ b/cpworker/src/output_zmq.h @@ -46,7 +46,9 @@ typedef struct // Heartbeat records in the batch. They are not captured packets, so a flush keeps // them out of the fwd/error_drop counters and counts them in heartbeat_packets. uint16_t heartbeat_num; - uint32_t heartbeat_bytes; + // Counted length of the captured packets in the batch: their sliced frame bytes, + // without the record header and MPLS label. A flush adds it to fwd/error_drop bytes. + uint32_t data_bytes; } zmq_pkts_buf_t; typedef struct ZmqOptions diff --git a/cpworker/tests/unit/CMakeLists.txt b/cpworker/tests/unit/CMakeLists.txt index 986205f6..c758c362 100644 --- a/cpworker/tests/unit/CMakeLists.txt +++ b/cpworker/tests/unit/CMakeLists.txt @@ -31,6 +31,7 @@ set(unity_tests misc output_file output_rotating_file + output_vxlan output_zmq packet_split req_pattern diff --git a/cpworker/tests/unit/output_vxlan.c b/cpworker/tests/unit/output_vxlan.c new file mode 100644 index 00000000..81f28c94 --- /dev/null +++ b/cpworker/tests/unit/output_vxlan.c @@ -0,0 +1,240 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "unity/src/unity.h" + +#include "output.h" +#include "output_gre.h" +#include "output_vxlan.h" +#include "pkt_dir.h" + +/* #308: every output counts each packet it is handed once, in exactly one of fwd / + * direction_drop / ratelimit_drop / error_drop, with its length after slice and without + * any encapsulation. */ + +#define FRAME_HDRS_LEN (14 + 20 + 20) + +static int receiver_fd; +static uint16_t receiver_port; +static output_stats_t stats; + +void setUp(void) +{ + memset(&stats, 0, sizeof(stats)); + + receiver_fd = socket(AF_INET, SOCK_DGRAM, 0); + TEST_ASSERT_NOT_EQUAL(-1, receiver_fd); + struct sockaddr_in addr; + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + TEST_ASSERT_EQUAL_INT(0, bind(receiver_fd, (struct sockaddr *)&addr, sizeof(addr))); + socklen_t len = sizeof(addr); + TEST_ASSERT_EQUAL_INT(0, getsockname(receiver_fd, (struct sockaddr *)&addr, &len)); + receiver_port = ntohs(addr.sin_port); + + struct timeval tv = {.tv_sec = 1, .tv_usec = 0}; + setsockopt(receiver_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); +} + +void tearDown(void) { close(receiver_fd); } + +static vxlan_output_t *new_vxlan(vxlan_options_t opts) +{ + opts.host = "127.0.0.1"; + opts.port = receiver_port; + opts.vni_version = 2; + opts.vni = 100; + opts.pmtudisc = -1; + char errbuf[ERROR_BUFFER_SIZE] = {0}; + vxlan_output_t *output = vxlan_output_new(opts, &stats, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + return output; +} + +// Ethernet + IPv4 + TCP frame of `len` bytes, with a consistent IP total length +static void build_tcp_frame(uint8_t *frame, uint32_t len) +{ + memset(frame, 0x5a, len); + frame[12] = 0x08; + frame[13] = 0x00; + + uint8_t *ip = frame + 14; + ip[0] = 0x45; + ip[1] = 0; + uint16_t tot_len = htons((uint16_t)(len - 14)); + memcpy(ip + 2, &tot_len, 2); + ip[6] = 0x40; // DF, offset 0 + ip[7] = 0; + ip[8] = 64; + ip[9] = 6; // TCP + + uint8_t *tcp = ip + 20; + tcp[12] = 0x50; // data offset 5 +} + +static int send_at(output_base_t *output, const uint8_t *frame, uint32_t caplen, int direct, time_t sec) +{ + struct pcap_pkthdr hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.ts.tv_sec = sec; + hdr.caplen = caplen; + hdr.len = caplen; + return output_send_packet(output, &hdr, frame, direct); +} + +static int recv_count(void) +{ + static uint8_t buf[70000]; + int n = 0; + while (recv(receiver_fd, buf, sizeof(buf), 0) > 0) + n++; + return n; +} + +static uint64_t bucket_packets(void) +{ + return stats.fwd_packets.packets + stats.direction_drop_packets.packets + stats.ratelimit_drop_packets.packets + + stats.error_drop_packets.packets; +} + +static uint64_t bucket_bytes(void) +{ + return stats.fwd_bytes.bytes + stats.direction_drop_bytes.bytes + stats.ratelimit_drop_bytes.bytes + + stats.error_drop_bytes.bytes; +} + +void test_vxlan_counts_sliced_length_without_header_or_capture_time(void) +{ + vxlan_options_t opts; + memset(&opts, 0, sizeof(opts)); + opts.capture_time = true; + opts.slice = 60; + vxlan_output_t *output = new_vxlan(opts); + + static uint8_t frame[200]; + build_tcp_frame(frame, sizeof(frame)); + TEST_ASSERT_EQUAL_INT(0, send_at(&output->base, frame, sizeof(frame), PKT_DIR_INCOMING, 1)); + TEST_ASSERT_EQUAL_INT(0, send_at(&output->base, frame, 50, PKT_DIR_INCOMING, 1)); + TEST_ASSERT_EQUAL_INT(-1, send_at(&output->base, frame, sizeof(frame), PKT_DIR_UNKNOWN, 1)); + vxlan_output_destroy(&output->base); + + TEST_ASSERT_EQUAL_INT(2, recv_count()); + TEST_ASSERT_EQUAL_UINT64(2, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(60 + 50, stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1, stats.direction_drop_packets.packets); + TEST_ASSERT_EQUAL_UINT64(60, stats.direction_drop_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(3, bucket_packets()); + TEST_ASSERT_EQUAL_UINT64(60 + 50 + 60, bucket_bytes()); +} + +void test_vxlan_split_packet_counts_once(void) +{ + vxlan_options_t opts; + memset(&opts, 0, sizeof(opts)); + opts.split.max_payload_size = 300; + vxlan_output_t *output = new_vxlan(opts); + + static uint8_t frame[FRAME_HDRS_LEN + 1000]; + build_tcp_frame(frame, sizeof(frame)); + TEST_ASSERT_EQUAL_INT(0, send_at(&output->base, frame, sizeof(frame), PKT_DIR_INCOMING, 1)); + vxlan_output_destroy(&output->base); + + TEST_ASSERT_EQUAL_INT(4, recv_count()); + TEST_ASSERT_EQUAL_UINT64(1, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(sizeof(frame), stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1, bucket_packets()); +} + +void test_vxlan_send_error_counts_whole_packet_as_error_drop(void) +{ + // A 65535-byte frame plus the VXLAN header exceeds the largest IPv4 UDP datagram, so + // sendto fails with EMSGSIZE + vxlan_options_t opts; + memset(&opts, 0, sizeof(opts)); + vxlan_output_t *output = new_vxlan(opts); + + static uint8_t frame[65535]; + memset(frame, 0x5a, sizeof(frame)); + TEST_ASSERT_EQUAL_INT(-1, send_at(&output->base, frame, sizeof(frame), PKT_DIR_INCOMING, 1)); + vxlan_output_destroy(&output->base); + + TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1, stats.error_drop_packets.packets); + TEST_ASSERT_EQUAL_UINT64(sizeof(frame), stats.error_drop_bytes.bytes); +} + +void test_vxlan_rate_limit_charges_sliced_length(void) +{ + // 1 Mbit/s gives a 125000-byte bucket. 125 frames sliced to 1000 bytes fit exactly; + // charging the 8-byte VXLAN header as well would reject the 125th. + vxlan_options_t opts; + memset(&opts, 0, sizeof(opts)); + opts.rate_limit_mbps = 1; + opts.slice = 1000; + vxlan_output_t *output = new_vxlan(opts); + + static uint8_t frame[1500]; + build_tcp_frame(frame, sizeof(frame)); + for (int i = 0; i < 125; i++) + TEST_ASSERT_EQUAL_INT(0, send_at(&output->base, frame, sizeof(frame), PKT_DIR_INCOMING, 1)); + TEST_ASSERT_EQUAL_INT(-1, send_at(&output->base, frame, sizeof(frame), PKT_DIR_INCOMING, 1)); + vxlan_output_destroy(&output->base); + + TEST_ASSERT_EQUAL_UINT64(125, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(125000, stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1, stats.ratelimit_drop_packets.packets); + TEST_ASSERT_EQUAL_UINT64(1000, stats.ratelimit_drop_bytes.bytes); +} + +void test_gre_counts_sliced_length_without_header(void) +{ + gre_options_t opts; + memset(&opts, 0, sizeof(opts)); + opts.host = "127.0.0.1"; + opts.service_tag = 1; + opts.pmtudisc = -1; + opts.rate_limit_mbps = 1; + opts.slice = 1000; + char errbuf[ERROR_BUFFER_SIZE] = {0}; + gre_output_t *output = gre_output_new(opts, &stats, errbuf); + if (!output) + TEST_IGNORE_MESSAGE("raw GRE socket unavailable (needs CAP_NET_RAW)"); + + // Same bucket arithmetic as the VXLAN test: the GRE header must not be charged + static uint8_t frame[1500]; + build_tcp_frame(frame, sizeof(frame)); + for (int i = 0; i < 125; i++) + TEST_ASSERT_EQUAL_INT(0, send_at(&output->base, frame, sizeof(frame), PKT_DIR_INCOMING, 1)); + TEST_ASSERT_EQUAL_INT(-1, send_at(&output->base, frame, sizeof(frame), PKT_DIR_INCOMING, 1)); + TEST_ASSERT_EQUAL_INT(-1, send_at(&output->base, frame, sizeof(frame), PKT_DIR_UNKNOWN, 1)); + gre_output_destroy(&output->base); + + TEST_ASSERT_EQUAL_UINT64(125, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(125000, stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1000, stats.ratelimit_drop_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1000, stats.direction_drop_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(127, bucket_packets()); +} + +int main(void) +{ + UNITY_BEGIN(); + + RUN_TEST(test_vxlan_counts_sliced_length_without_header_or_capture_time); + RUN_TEST(test_vxlan_split_packet_counts_once); + RUN_TEST(test_vxlan_send_error_counts_whole_packet_as_error_drop); + RUN_TEST(test_vxlan_rate_limit_charges_sliced_length); + RUN_TEST(test_gre_counts_sliced_length_without_header); + + return UNITY_END(); +} diff --git a/cpworker/tests/unit/output_zmq.c b/cpworker/tests/unit/output_zmq.c index 04e5836b..a4db13ae 100644 --- a/cpworker/tests/unit/output_zmq.c +++ b/cpworker/tests/unit/output_zmq.c @@ -410,12 +410,11 @@ void test_heartbeat_in_batch_with_data_counts_only_the_data(void) send_frame_at(output, now - 1); zmq_heartbeat(&output->base, make_heartbeat_due(output)); - int n = recv_batch(); - TEST_ASSERT_GREATER_THAN_INT(0, n); + TEST_ASSERT_GREATER_THAN_INT(0, recv_batch()); TEST_ASSERT_EQUAL_UINT16(2, batch_pkts_num()); TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets); TEST_ASSERT_EQUAL_UINT64(1, stats.fwd_packets.packets); - TEST_ASSERT_EQUAL_UINT64((uint64_t)n - HEARTBEAT_RECORD_SIZE, stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(64, stats.fwd_bytes.bytes); zmq_output_destroy(&output->base); } @@ -451,18 +450,101 @@ void test_failed_batch_with_heartbeat_drops_only_the_data(void) time_t now = time(NULL); send_frame_at(output, now - 1); - uint32_t batch_size = output->pkts_buf.batch_bufpos + HEARTBEAT_RECORD_SIZE; zmq_heartbeat(&output->base, make_heartbeat_due(output)); TEST_ASSERT_EQUAL_UINT64(1, output->error_info.nb_drop_batches); TEST_ASSERT_EQUAL_UINT64(1, stats.heartbeat_packets.packets); TEST_ASSERT_EQUAL_UINT64(1, stats.error_drop_packets.packets); - TEST_ASSERT_EQUAL_UINT64(batch_size - HEARTBEAT_RECORD_SIZE, stats.error_drop_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(64, stats.error_drop_bytes.bytes); TEST_ASSERT_EQUAL_UINT64(0, stats.fwd_packets.packets); destroy_without_linger(output); } +/* ---- #308: counters count each packet once, with its sliced frame length ---- */ + +static uint64_t bucket_packets(void) +{ + return stats.fwd_packets.packets + stats.direction_drop_packets.packets + stats.ratelimit_drop_packets.packets + + stats.error_drop_packets.packets; +} + +static uint64_t bucket_bytes(void) +{ + return stats.fwd_bytes.bytes + stats.direction_drop_bytes.bytes + stats.ratelimit_drop_bytes.bytes + + stats.error_drop_bytes.bytes; +} + +static int send_frame_dir(zmq_output_t *output, const uint8_t *frame, uint32_t caplen, int direct) +{ + struct pcap_pkthdr hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.ts.tv_sec = 1; + hdr.caplen = caplen; + hdr.len = caplen; + return output_send_packet(&output->base, &hdr, frame, direct); +} + +void test_counters_use_sliced_frame_length_without_headers(void) +{ + char errbuf[ERROR_BUFFER_SIZE] = {0}; + zmq_output_t *output = new_output(40, (char *)VALID_UUID, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + + uint8_t frame[100]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x08; + frame[13] = 0x00; + + TEST_ASSERT_EQUAL_INT(0, send_frame_dir(output, frame, sizeof(frame), PKT_DIR_INCOMING)); + TEST_ASSERT_EQUAL_INT(0, send_frame_dir(output, frame, 30, PKT_DIR_OUTGOING)); + TEST_ASSERT_EQUAL_INT(-1, send_frame_dir(output, frame, sizeof(frame), PKT_DIR_UNKNOWN)); + zmq_output_destroy(&output->base); + + TEST_ASSERT_GREATER_THAN_INT(0, recv_batch()); + TEST_ASSERT_EQUAL_UINT64(2, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(40 + 30, stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1, stats.direction_drop_packets.packets); + TEST_ASSERT_EQUAL_UINT64(40, stats.direction_drop_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(3, bucket_packets()); + TEST_ASSERT_EQUAL_UINT64(40 + 30 + 40, bucket_bytes()); +} + +void test_rate_limit_charges_and_counts_sliced_frame_length(void) +{ + // 1 Mbit/s gives a 125000-byte bucket. 125 frames sliced to 1000 bytes fit exactly; one + // more is rejected. Charging the 4-byte MPLS label as well would reject the 125th. + zmq_options_t opts = { + .host = "127.0.0.1", + .port = receiver_port, + .hwm = 100, + .service_tag = 1, + .uuid = (char *)VALID_UUID, + .rate_limit_mbps = 1, + .slice = 1000, + .heartbeat_ms = 0, + }; + char errbuf[ERROR_BUFFER_SIZE] = {0}; + zmq_output_t *output = zmq_output_new(opts, &stats, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + + static uint8_t frame[1500]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x08; + frame[13] = 0x00; + + for (int i = 0; i < 125; i++) + TEST_ASSERT_EQUAL_INT(0, send_frame_dir(output, frame, sizeof(frame), PKT_DIR_INCOMING)); + TEST_ASSERT_EQUAL_INT(-1, send_frame_dir(output, frame, sizeof(frame), PKT_DIR_INCOMING)); + zmq_output_destroy(&output->base); + + TEST_ASSERT_GREATER_THAN_INT(0, recv_batch()); + TEST_ASSERT_EQUAL_UINT64(125, stats.fwd_packets.packets); + TEST_ASSERT_EQUAL_UINT64(125000, stats.fwd_bytes.bytes); + TEST_ASSERT_EQUAL_UINT64(1, stats.ratelimit_drop_packets.packets); + TEST_ASSERT_EQUAL_UINT64(1000, stats.ratelimit_drop_bytes.bytes); +} + int main(void) { UNITY_BEGIN(); @@ -488,5 +570,8 @@ int main(void) RUN_TEST(test_failed_heartbeat_is_not_counted); RUN_TEST(test_failed_batch_with_heartbeat_drops_only_the_data); + RUN_TEST(test_counters_use_sliced_frame_length_without_headers); + RUN_TEST(test_rate_limit_charges_and_counts_sliced_frame_length); + return UNITY_END(); } diff --git a/docs/REFACTOR-CPCTL.md b/docs/REFACTOR-CPCTL.md index 24415b62..a3b0d6f8 100644 --- a/docs/REFACTOR-CPCTL.md +++ b/docs/REFACTOR-CPCTL.md @@ -345,6 +345,39 @@ Sampling and timestamps: - `counters.heartbeat_packets` counts the zmq heartbeats sent. Heartbeats are not captured packets and are not part of `fwd_packets` / `fwd_bytes`. +Output counters (`fwd_*`, `direction_drop_*`, `ratelimit_drop_*`, +`error_drop_*`) mean the same for every output type: + +- Each packet the capturer hands to an output is counted once, as 1 packet + and `L` bytes, in exactly one of the four buckets, once its outcome is known. +- `L` is the packet's `caplen` after the output's `slice`, without any + encapsulation: GRE/VXLAN headers, the VXLAN `capture_time` trailer, ZMQ + batch and record headers, the ZMQ MPLS label and outer IP/UDP headers are + not counted. For zmq, `slice` keeps the Ethernet header and VLAN tag stack + (ZMQ-WIRE-FORMAT.md §4.1), so `L` is the record's frame bytes. + +| Bucket | The packet | +| --- | --- | +| `direction_drop` | has an unknown direction and is dropped | +| `ratelimit_drop` | is rejected by `rate_limit_mbps` | +| `error_drop` | is not delivered in full: send error, partial send, a failed split fragment, a failed zmq batch, a file write error, or a frame too short to send | +| `fwd` | is delivered in full: every split fragment sent, its zmq batch sent, or written to the file | + +So, for packets and for bytes: + +``` +fwd + direction_drop + ratelimit_drop + error_drop = packets handed to outputs +``` + +- A packet split into several VXLAN fragments counts once. +- A packet in a zmq batch that has not been sent yet is in no bucket. +- `rate_limit_mbps` charges the same `L` per packet, once per packet, so while + the limit is active `fwd_bytes` grows by at most `rate_limit_mbps`. +- The counters are summed over every output of every task. With two outputs + per task, the right-hand side above is about twice `cap_packets`. +- `cap_bytes` is the unsliced `caplen`, so with `slice` set `fwd_bytes` is + lower than `cap_bytes`. + ### 6.3 info Minimum useful set, all derivable from data cpworker already has: diff --git a/docs/USAGE-CPWORKER-zh-Hans.md b/docs/USAGE-CPWORKER-zh-Hans.md index e7d8e8fe..87957700 100644 --- a/docs/USAGE-CPWORKER-zh-Hans.md +++ b/docs/USAGE-CPWORKER-zh-Hans.md @@ -118,7 +118,7 @@ cpworker 是基于 libpcap 的网络抓包工具，支持多种输出方式和 | 参数 | 类型 | 默认值 | 说明 | |-------------------|---------|---------|-----| | type | string | - | 输出类型 | -| rate_limit_mbps | int | 0 | 最大输出速率，单位 Mbps（>= 0）；0 表示不限速 | +| rate_limit_mbps | int | 0 | 最大输出速率，单位 Mbps（>= 0）；0 表示不限速。每个报文按 `slice` 之后的长度计费，不含封装头，与 `fwd_bytes` 的计数字节一致 | | slice | int | 0 | 输出时数据包截断长度，单位字节（>= 0）；0 表示不截断 | ## output.gre 参数列表 diff --git a/docs/USAGE-CPWORKER.md b/docs/USAGE-CPWORKER.md index 2244bfbd..fb20d5ad 100644 --- a/docs/USAGE-CPWORKER.md +++ b/docs/USAGE-CPWORKER.md @@ -117,7 +117,7 @@ Values shown in hexadecimal (e.g. `0xFFF`) must be written in decimal in the JSO | Parameter | Type | Default | Description | |-------------------|----------|---------|-------------| | type | string | - | Output type | -| rate_limit_mbps | int | 0 | Max output rate in Mbps (>= 0); 0 means unlimited | +| rate_limit_mbps | int | 0 | Max output rate in Mbps (>= 0); 0 means unlimited. Each packet is charged its length after `slice`, without encapsulation headers, the same bytes `fwd_bytes` counts | | slice | int | 0 | Packet truncation size in bytes (>= 0); 0 means no truncation | ## output.gre Parameters diff --git a/docs/ZMQ-WIRE-FORMAT-zh-Hans.md b/docs/ZMQ-WIRE-FORMAT-zh-Hans.md index 74a13378..54b8c19b 100644 --- a/docs/ZMQ-WIRE-FORMAT-zh-Hans.md +++ b/docs/ZMQ-WIRE-FORMAT-zh-Hans.md @@ -216,10 +216,10 @@ heartbeat 记录不是抓到的报文，输出计数器不计入它： | batch 内容 | 发送成功 | 发送失败 | | --- | --- | --- | -| 有抓到的报文（可能夹带 heartbeat） | 数据记录和 batch header → `fwd_packets` / `fwd_bytes`；heartbeat → `heartbeat_packets` | 数据记录和 batch header → `error_drop_packets` / `error_drop_bytes` | +| 有抓到的报文（可能夹带 heartbeat） | 抓到的报文 → `fwd_packets` / `fwd_bytes`；heartbeat → `heartbeat_packets` | 抓到的报文 → `error_drop_packets` / `error_drop_bytes` | | 只有 heartbeat | `heartbeat_packets` | 不计数 | -字节数为 batch 字节数减去 heartbeat 记录。丢弃的 batch 个数只出现在 worker 的周期性错误日志里（`nb_drop_batches`），没有对应的统计计数器。 +每个抓到的报文按 `slice` 之后的帧字节数计入（即记录的 `caplen` 减去 4 字节 MPLS 标签），batch header 和记录头不计入。各类输出计数器的统一定义见 docs/REFACTOR-CPCTL.md §6.2。丢弃的 batch 个数只出现在 worker 的周期性错误日志里（`nb_drop_batches`），没有对应的统计计数器。 --- diff --git a/docs/ZMQ-WIRE-FORMAT.md b/docs/ZMQ-WIRE-FORMAT.md index 55311a9a..cbe3d0b9 100644 --- a/docs/ZMQ-WIRE-FORMAT.md +++ b/docs/ZMQ-WIRE-FORMAT.md @@ -235,10 +235,12 @@ Heartbeat records are not captured packets, so the output counters leave them ou | Batch | Sent | Dropped | | --- | --- | --- | -| Captured packets (and heartbeats, if any) | data records and batch header → `fwd_packets` / `fwd_bytes`; heartbeats → `heartbeat_packets` | data records and batch header → `error_drop_packets` / `error_drop_bytes` | +| Captured packets (and heartbeats, if any) | captured packets → `fwd_packets` / `fwd_bytes`; heartbeats → `heartbeat_packets` | captured packets → `error_drop_packets` / `error_drop_bytes` | | Heartbeats only | `heartbeat_packets` | not counted | -Bytes are the batch bytes minus the heartbeat records. The worker's periodic error log reports +Each captured packet adds its frame bytes after `slice` (the record's `caplen` minus the 4-byte +MPLS label). The batch header and record headers are not counted. docs/REFACTOR-CPCTL.md §6.2 +defines the output counters for every output type. The worker's periodic error log reports the number of dropped batches (`nb_drop_batches`); no stats counter carries it. --- From a928e3b91db087fabf10b4152160d75dc63b8208 Mon Sep 17 00:00:00 2001 From: "timmy.yuan" Date: Fri, 9 Oct 2026 10:45:09 +0800 Subject: [PATCH 2/2] refactor(cpworker): name the counted frame length frame_len in every output length and caplen meant different things in each output: the sliced frame, the frame capped at 65535, the VXLAN payload with capture_time, or the zmq record length with the MPLS label. send_whole_or_split took both caplen and length, and parsed the split with the uncapped one. output_frame_len() in output.h now computes the length an output carries and counts (caplen after slice, capped per output), and gre, vxlan, null and zmq use it as frame_len. header->caplen always means the captured length. vxlan uses payload_len for the frame plus capture_time, and zmq record_len for the frame plus the MPLS label. vxlan parses and sends a split packet with the same frame_len. --- cpworker/src/output.h | 10 +++++++ cpworker/src/output_gre.c | 26 +++++++----------- cpworker/src/output_gre.h | 3 ++- cpworker/src/output_null.c | 15 +++++------ cpworker/src/output_vxlan.c | 53 +++++++++++++++++-------------------- cpworker/src/output_vxlan.h | 5 ++-- cpworker/src/output_zmq.c | 34 ++++++++++++------------ cpworker/src/output_zmq.h | 2 ++ 8 files changed, 75 insertions(+), 73 deletions(-) diff --git a/cpworker/src/output.h b/cpworker/src/output.h index a123747d..2e09cb0b 100644 --- a/cpworker/src/output.h +++ b/cpworker/src/output.h @@ -27,6 +27,16 @@ typedef struct OutputEntry OutputFactory factory; } output_entry_t; +// Length of the frame an output carries and counts: the captured length after slice, capped at +// the largest frame the output can carry. Every output counter and rate limiter uses this length. +static inline size_t output_frame_len(uint32_t caplen, int slice, size_t max_len) +{ + size_t frame_len = caplen; + if (slice > 0 && (uint32_t)slice < caplen) + frame_len = (size_t)slice; + return frame_len < max_len ? frame_len : max_len; +} + static inline int output_send_packet(output_base_t *output, const struct pcap_pkthdr *header, const uint8_t *pkt_data, int direct) { diff --git a/cpworker/src/output_gre.c b/cpworker/src/output_gre.c index 215d5ec5..1de90a48 100644 --- a/cpworker/src/output_gre.c +++ b/cpworker/src/output_gre.c @@ -42,26 +42,20 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const { gre_output_t *output = (gre_output_t *)self; - int32_t caplen = header->caplen; - if (output->slice > 0 && output->slice < caplen) - { - caplen = output->slice; - } - - size_t length = (size_t)(caplen <= 65535 ? caplen : 65535); + const size_t frame_len = output_frame_len(header->caplen, output->slice, GRE_MAX_FRAME_LEN); if (direct == PKT_DIR_UNKNOWN) { - bytes_stats_add(&output->base.stats->direction_drop_bytes, length); + bytes_stats_add(&output->base.stats->direction_drop_bytes, frame_len); packets_stats_add(&output->base.stats->direction_drop_packets, 1); return -1; } if (output->rate_limit_mbps > 0) { - if (!token_bucket_consume(&output->throttle, length, header->ts)) + if (!token_bucket_consume(&output->throttle, frame_len, header->ts)) { - bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, length); + bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, frame_len); packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1); return -1; } @@ -69,7 +63,7 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const struct gre_header *gre_hdr = (struct gre_header *)output->buf; gre_hdr->keybit = htonl(output->service_tag | (direct << 28)); - memcpy(&(output->buf[GRE_HEADER_LEN]), pkt_data, length); + memcpy(&(output->buf[GRE_HEADER_LEN]), pkt_data, frame_len); // check error_info if (output->error_info.first_pktsec == 0) @@ -84,7 +78,7 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const const int max_retries = 10; do { - ssize_t send_bytes = sendto(output->socket_fd, output->buf, GRE_HEADER_LEN + length, 0, + ssize_t send_bytes = sendto(output->socket_fd, output->buf, GRE_HEADER_LEN + frame_len, 0, (struct sockaddr *)&output->remote_addr, sizeof(struct sockaddr_in)); if (send_bytes == -1) @@ -112,21 +106,21 @@ int gre_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const output->error_info.nb_other_send_error_drops++; } - bytes_stats_add(&output->base.stats->error_drop_bytes, length); + bytes_stats_add(&output->base.stats->error_drop_bytes, frame_len); packets_stats_add(&output->base.stats->error_drop_packets, 1); return -1; } - if (send_bytes < GRE_HEADER_LEN + length) + if (send_bytes < GRE_HEADER_LEN + frame_len) { output->error_info.nb_partial_sends++; - bytes_stats_add(&output->base.stats->error_drop_bytes, length); + bytes_stats_add(&output->base.stats->error_drop_bytes, frame_len); packets_stats_add(&output->base.stats->error_drop_packets, 1); return -1; } - bytes_stats_add(&output->base.stats->fwd_bytes, length); + bytes_stats_add(&output->base.stats->fwd_bytes, frame_len); packets_stats_add(&output->base.stats->fwd_packets, 1); return 0; } while (true); diff --git a/cpworker/src/output_gre.h b/cpworker/src/output_gre.h index 19277ed1..2df124eb 100644 --- a/cpworker/src/output_gre.h +++ b/cpworker/src/output_gre.h @@ -9,7 +9,8 @@ #include "output.h" #include "ratelimit.h" -#define GRE_OUTPUT_BUFSIZE 65551 // 8(GRE_HEADER_LEN) + 65535 +#define GRE_MAX_FRAME_LEN 65535 +#define GRE_OUTPUT_BUFSIZE 65551 // 8(GRE_HEADER_LEN) + 65535(GRE_MAX_FRAME_LEN) typedef struct GreOptions { diff --git a/cpworker/src/output_null.c b/cpworker/src/output_null.c index 5e646d02..71f097e2 100644 --- a/cpworker/src/output_null.c +++ b/cpworker/src/output_null.c @@ -1,3 +1,4 @@ +#include #include #include "errorf.h" @@ -10,29 +11,25 @@ int null_send_packet(output_base_t *self, const struct pcap_pkthdr *header, cons { null_output_t *output = (null_output_t *)self; - int32_t length = header->caplen; - if (output->slice > 0 && output->slice < length) - { - length = output->slice; - } + const size_t frame_len = output_frame_len(header->caplen, output->slice, SIZE_MAX); if (direct == PKT_DIR_UNKNOWN) { - bytes_stats_add(&output->base.stats->direction_drop_bytes, length); + bytes_stats_add(&output->base.stats->direction_drop_bytes, frame_len); packets_stats_add(&output->base.stats->direction_drop_packets, 1); return -1; } if (output->rate_limit_mbps > 0) { - if (!token_bucket_consume(&output->throttle, length, header->ts)) + if (!token_bucket_consume(&output->throttle, frame_len, header->ts)) { - bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, length); + bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, frame_len); packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1); return -1; } } - bytes_stats_add(&output->base.stats->fwd_bytes, length); + bytes_stats_add(&output->base.stats->fwd_bytes, frame_len); packets_stats_add(&output->base.stats->fwd_packets, 1); return 0; } diff --git a/cpworker/src/output_vxlan.c b/cpworker/src/output_vxlan.c index f24c66e6..90c6ae0e 100644 --- a/cpworker/src/output_vxlan.c +++ b/cpworker/src/output_vxlan.c @@ -94,21 +94,24 @@ static void flush_error_info(vxlan_output_t *output) } // Sends one VXLAN datagram. Returns 0 when it was sent in full, -1 otherwise. The caller does the counting. -static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const uint8_t *pkt_data, size_t length, +static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const uint8_t *pkt_data, size_t frame_len, int direct) { struct vxlan_header *vxlan_hdr = (struct vxlan_header *)output->buf; - memcpy(&(output->buf[VXLAN_HEADER_LEN]), pkt_data, length); + memcpy(&(output->buf[VXLAN_HEADER_LEN]), pkt_data, frame_len); + + // VXLAN payload: the frame, followed by the capture time when enabled + size_t payload_len = frame_len; if (output->capture_time) { uint32_t tv_sec = htonl(ts.tv_sec); // 注意：通过libpcap获取的捕获时间精度为微秒，而数据包中附加的时间为纳秒，所以需要*1000 uint32_t tv_nsec = htonl(ts.tv_usec * 1000); - memcpy(&(output->buf[VXLAN_HEADER_LEN + length]), &tv_sec, 4); - length += 4; - memcpy(&(output->buf[VXLAN_HEADER_LEN + length]), &tv_nsec, 4); - length += 4; + memcpy(&(output->buf[VXLAN_HEADER_LEN + payload_len]), &tv_sec, 4); + payload_len += 4; + memcpy(&(output->buf[VXLAN_HEADER_LEN + payload_len]), &tv_nsec, 4); + payload_len += 4; } if (output->vni_version == 1) { @@ -133,7 +136,7 @@ static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const const int max_retries = 10; do { - ssize_t send_bytes = sendto(output->socket_fd, output->buf, VXLAN_HEADER_LEN + length, 0, + ssize_t send_bytes = sendto(output->socket_fd, output->buf, VXLAN_HEADER_LEN + payload_len, 0, (struct sockaddr *)&output->remote_addr, sizeof(struct sockaddr_in)); if (send_bytes == -1) @@ -163,7 +166,7 @@ static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const return -1; } - if (send_bytes < VXLAN_HEADER_LEN + length) + if (send_bytes < VXLAN_HEADER_LEN + payload_len) { output->error_info.nb_partial_sends++; return -1; @@ -175,19 +178,19 @@ static int do_send_packet(vxlan_output_t *output, const struct timeval ts, const // Sends the packet, split into fragments when configured. Returns 0 only when every datagram was sent in full. static int send_whole_or_split(vxlan_output_t *output, const struct pcap_pkthdr *header, const uint8_t *pkt_data, - int32_t caplen, size_t length, int direct) + size_t frame_len, int direct) { // Fast path - if (output->split.max_payload_size <= 0 || length <= (size_t)output->split.max_payload_size) - return do_send_packet(output, header->ts, pkt_data, length, direct); + if (output->split.max_payload_size <= 0 || frame_len <= (size_t)output->split.max_payload_size) + return do_send_packet(output, header->ts, pkt_data, frame_len, direct); packet_parse_result_t parse_result; - if (!parse_packet(pkt_data, caplen, &parse_result)) - return do_send_packet(output, header->ts, pkt_data, length, direct); + if (!parse_packet(pkt_data, frame_len, &parse_result)) + return do_send_packet(output, header->ts, pkt_data, frame_len, direct); int fragment_count = calculate_fragment_count(&parse_result, output->split.max_payload_size); if (fragment_count == 1) - return do_send_packet(output, header->ts, pkt_data, length, direct); + return do_send_packet(output, header->ts, pkt_data, frame_len, direct); for (int i = 0; i < fragment_count; i++) { @@ -206,26 +209,20 @@ int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, con { vxlan_output_t *output = (vxlan_output_t *)self; - int32_t caplen = header->caplen; - if (output->slice > 0 && output->slice < caplen) - { - caplen = output->slice; - } - - size_t length = (size_t)(caplen <= 65535 ? caplen : 65535); + const size_t frame_len = output_frame_len(header->caplen, output->slice, VXLAN_MAX_FRAME_LEN); if (direct == PKT_DIR_UNKNOWN) { - bytes_stats_add(&output->base.stats->direction_drop_bytes, length); + bytes_stats_add(&output->base.stats->direction_drop_bytes, frame_len); packets_stats_add(&output->base.stats->direction_drop_packets, 1); return -1; } if (output->rate_limit_mbps > 0) { - if (!token_bucket_consume(&output->throttle, length, header->ts)) + if (!token_bucket_consume(&output->throttle, frame_len, header->ts)) { - bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, length); + bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, frame_len); packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1); return -1; } @@ -240,15 +237,15 @@ int vxlan_send_packet(output_base_t *self, const struct pcap_pkthdr *header, con output->error_info.first_pktsec = header->ts.tv_sec; } - // A packet counts once, with its sliced length, whether or not it was split - if (send_whole_or_split(output, header, pkt_data, caplen, length, direct) != 0) + // A packet counts once, with its frame_len, whether or not it was split + if (send_whole_or_split(output, header, pkt_data, frame_len, direct) != 0) { - bytes_stats_add(&output->base.stats->error_drop_bytes, length); + bytes_stats_add(&output->base.stats->error_drop_bytes, frame_len); packets_stats_add(&output->base.stats->error_drop_packets, 1); return -1; } - bytes_stats_add(&output->base.stats->fwd_bytes, length); + bytes_stats_add(&output->base.stats->fwd_bytes, frame_len); packets_stats_add(&output->base.stats->fwd_packets, 1); return 0; } diff --git a/cpworker/src/output_vxlan.h b/cpworker/src/output_vxlan.h index 951f7ed5..44eadc5a 100644 --- a/cpworker/src/output_vxlan.h +++ b/cpworker/src/output_vxlan.h @@ -11,7 +11,8 @@ #include "output.h" #include "ratelimit.h" -#define VXLAN_OUTPUT_BUFSIZE 65551 // 8(VXLAN_HEADER_LEN) + 65535 + 8(capture_time) +#define VXLAN_MAX_FRAME_LEN 65535 +#define VXLAN_OUTPUT_BUFSIZE 65551 // 8(VXLAN_HEADER_LEN) + 65535(VXLAN_MAX_FRAME_LEN) + 8(capture_time) typedef struct VxlanOptions { @@ -29,7 +30,7 @@ typedef struct VxlanOptions struct { uint16_t max_payload_size; - bool recalculate_checksum; // Whether to recalculate checksums after splitting (default: false) + bool recalculate_checksum; // Whether to recalculate checksums after splitting (default: false) } split; } vxlan_options_t; diff --git a/cpworker/src/output_zmq.c b/cpworker/src/output_zmq.c index 59d745a1..d42b501b 100644 --- a/cpworker/src/output_zmq.c +++ b/cpworker/src/output_zmq.c @@ -178,16 +178,14 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const { zmq_output_t *output = (zmq_output_t *)self; - // The record length is a uint16 that also covers the MPLS header - uint32_t caplen = header->caplen <= 65531 ? header->caplen : 65531; - // The MPLS header goes after the innermost VLAN tag, so a frame whose Ethernet header - // and tag stack were not fully captured cannot form a record - const size_t l2_len = l2_header_len(pkt_data, caplen); + // and tag stack were not fully captured cannot form a record. Only the first + // ZMQ_MAX_FRAME_LEN bytes fit in a record. + const size_t l2_len = l2_header_len(pkt_data, output_frame_len(header->caplen, 0, ZMQ_MAX_FRAME_LEN)); if (l2_len == 0) { output->error_info.nb_too_small_packets++; - bytes_stats_add(&output->base.stats->error_drop_bytes, caplen); + bytes_stats_add(&output->base.stats->error_drop_bytes, header->caplen); packets_stats_add(&output->base.stats->error_drop_packets, 1); zmq_flush_if_stale(output, header->ts.tv_sec); @@ -195,24 +193,26 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const } // slice truncates the payload only: the Ethernet header and VLAN tags are always kept - if (output->slice > 0 && (uint32_t)output->slice < caplen) - caplen = (uint32_t)output->slice > l2_len ? (uint32_t)output->slice : l2_len; + size_t frame_len = output_frame_len(header->caplen, output->slice, ZMQ_MAX_FRAME_LEN); + if (frame_len < l2_len) + frame_len = l2_len; - uint16_t length = (uint16_t)caplen + sizeof(mpls_header); + // The record length is a uint16 that also covers the MPLS header + const uint16_t record_len = (uint16_t)(frame_len + sizeof(mpls_header)); zmq_pkts_buf_t *pkts_buf = &output->pkts_buf; if (direct == PKT_DIR_UNKNOWN) { - bytes_stats_add(&output->base.stats->direction_drop_bytes, caplen); + bytes_stats_add(&output->base.stats->direction_drop_bytes, frame_len); packets_stats_add(&output->base.stats->direction_drop_packets, 1); zmq_flush_if_stale(output, header->ts.tv_sec); return -1; } - if (output->rate_limit_mbps > 0 && !token_bucket_consume(&output->throttle, caplen, header->ts)) + if (output->rate_limit_mbps > 0 && !token_bucket_consume(&output->throttle, frame_len, header->ts)) { - bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, caplen); + bytes_stats_add(&output->base.stats->ratelimit_drop_bytes, frame_len); packets_stats_add(&output->base.stats->ratelimit_drop_packets, 1); zmq_flush_if_stale(output, header->ts.tv_sec); @@ -225,7 +225,7 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const zmq_pkt_hdr_t pkt_hdr = { .tv_sec = htonl((uint32_t)header->ts.tv_sec), .tv_usec = htonl((uint32_t)header->ts.tv_usec), - .caplen = htonl((uint32_t)length), + .caplen = htonl((uint32_t)record_len), .len = htonl((uint32_t)header->len + sizeof(mpls_header)), }; @@ -235,7 +235,7 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const (pkts_buf->first_pktsec != 0 && header->ts.tv_sec > pkts_buf->first_pktsec + ZMQ_PKTS_FLUSH_MAX_DUR_SEC); const bool is_buffer_full = - (pkts_buf->batch_bufpos + sizeof(length) + sizeof(pkt_hdr) + length > ZMQ_MAX_BATCH_BUF_SIZE); + (pkts_buf->batch_bufpos + sizeof(record_len) + sizeof(pkt_hdr) + record_len > ZMQ_MAX_BATCH_BUF_SIZE); if (is_pkt_num_exceeded || is_time_diff_exceeded || is_buffer_full) { @@ -246,7 +246,7 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const if (pkts_buf->first_pktsec == 0) pkts_buf->first_pktsec = header->ts.tv_sec; - uint16_t hlen = htons(length); + uint16_t hlen = htons(record_len); uint32_t buff_pos = pkts_buf->batch_bufpos; memcpy(&(pkts_buf->buf[buff_pos]), &hlen, ZMQ_PKT_DATA_LEN_SIZE); @@ -287,13 +287,13 @@ int zmq_send_packet(output_base_t *self, const struct pcap_pkthdr *header, const // Copy payload (everything after Ethernet + all VLANs) const size_t payload_offset = sizeof(struct ether_header) + vlan_total_size; - const size_t payload_copy_len = caplen - payload_offset; + const size_t payload_copy_len = frame_len - payload_offset; memcpy(&(pkts_buf->buf[buff_pos]), pkt_data + payload_offset, payload_copy_len); buff_pos += payload_copy_len; pkts_buf->batch_bufpos = buff_pos; pkts_buf->batch_hdr.pkts_num++; - pkts_buf->data_bytes += caplen; + pkts_buf->data_bytes += frame_len; output->last_pkt_tv.tv_sec = header->ts.tv_sec; output->last_pkt_tv.tv_usec = header->ts.tv_usec; diff --git a/cpworker/src/output_zmq.h b/cpworker/src/output_zmq.h index 1af3b9e9..14031d48 100644 --- a/cpworker/src/output_zmq.h +++ b/cpworker/src/output_zmq.h @@ -11,6 +11,8 @@ #include "output.h" #include "ratelimit.h" +// Largest frame in a record: the uint16 record length also covers the 4-byte MPLS header +#define ZMQ_MAX_FRAME_LEN 65531 #define ZMQ_MAX_BATCH_BUF_SIZE 1048576 // 1 * 1024 * 1024; #define ZMQ_PKTS_FLUSH_MAX_DUR_SEC 1 #define ZMQ_PKTS_FLUSH_MAX_NUM 65535