Merge pull request #1034 from LibreQoE/feat/packet_kprobe2

Feat/packet kprobe2
This commit is contained in:
Robert Chacón authored and GitHub committed 2026-04-07 17:24:50 -06:00
commit 8cecddee77
21 files changed
+540 -62

No files matched your search

+3
View File
@@ -207,6 +207,9 @@ pub struct Circuit {
pub ip: IpAddr,
/// Current bytes-per-second passing through this host.
pub bytes_per_second: DownUpOrder<u64>,
/// Current bytes-per-second actually transmitted for this host.
#[serde(default)]
pub actual_bytes_per_second: DownUpOrder<u64>,
/// Median latency for this host at the current time.
pub median_latency: Option<f32>,
/// Current RTT p50 (nanoseconds), per direction.
@@ -0,0 +1,171 @@
#pragma once
// Kprobe-only helpers.
// IMPORTANT: Do not use these from XDP/TC paths; keep XDP verifier balance intact.
// Minimal VLAN header representation for scanning 802.1Q/AD tags.
struct vlan_hdr_kprobe {
__be16 h_vlan_TCI;
__be16 h_vlan_encapsulated_proto;
};
static __always_inline __be16 kprobe_scan_current_vlan(struct sk_buff *skb)
{
unsigned char *data = (unsigned char *)BPF_CORE_READ(skb, data);
if (!data) {
return 0;
}
struct ethhdr eth = {0};
if (bpf_probe_read_kernel(&eth, sizeof(eth), data) < 0) {
return 0;
}
__u16 eth_type = bpf_ntohs(eth.h_proto);
__u32 offset = sizeof(struct ethhdr);
__be16 current_vlan = 0;
// Keep this very small and bounded.
#pragma unroll
for (int i = 0; i < 2; i++) {
if (eth_type != ETH_P_8021AD && eth_type != ETH_P_8021Q) {
break;
}
struct vlan_hdr_kprobe vlan = {0};
if (bpf_probe_read_kernel(&vlan, sizeof(vlan), data + offset) < 0) {
break;
}
current_vlan = vlan.h_vlan_TCI;
eth_type = bpf_ntohs(vlan.h_vlan_encapsulated_proto);
offset += sizeof(struct vlan_hdr_kprobe);
}
return current_vlan;
}
static __always_inline void kprobe_encode_ipv4(__be32 addr, struct in6_addr *out_address)
{
out_address->in6_u.u6_addr32[0] = (__u32)0xFFFFFFFF;
out_address->in6_u.u6_addr32[1] = (__u32)0xFFFFFFFF;
out_address->in6_u.u6_addr32[2] = (__u32)0xFFFFFFFF;
out_address->in6_u.u6_addr32[3] = addr;
}
static __always_inline void kprobe_encode_ipv6(struct in6_addr *ipv6_address, struct in6_addr *out_address)
{
__builtin_memcpy(&out_address->in6_u.u6_addr8, &ipv6_address->in6_u.u6_addr8, 16);
}
static __always_inline __be16 kprobe_current_vlan(struct sk_buff *skb)
{
// Prefer the skb metadata tag (matches TC egress behavior).
__u16 vlan_tci = BPF_CORE_READ(skb, vlan_tci);
if (vlan_tci) {
return bpf_htons(vlan_tci);
}
// Fall back to scanning the L2 header for non-accelerated VLAN tags.
return kprobe_scan_current_vlan(skb);
}
static __always_inline bool kprobe_build_host_key(
struct sk_buff *skb,
__u8 effective_direction,
struct in6_addr *host_key
)
{
unsigned char *data = (unsigned char *)BPF_CORE_READ(skb, data);
if (!data) {
return false;
}
struct ethhdr eth = {0};
if (bpf_probe_read_kernel(&eth, sizeof(eth), data) < 0) {
return false;
}
__u16 eth_type = bpf_ntohs(eth.h_proto);
__u32 offset = sizeof(struct ethhdr);
// Walk up to two VLAN tags.
#pragma unroll
for (int i = 0; i < 2; i++) {
if (eth_type != ETH_P_8021AD && eth_type != ETH_P_8021Q) {
break;
}
struct vlan_hdr_kprobe vlan = {0};
if (bpf_probe_read_kernel(&vlan, sizeof(vlan), data + offset) < 0) {
return false;
}
eth_type = bpf_ntohs(vlan.h_vlan_encapsulated_proto);
offset += sizeof(struct vlan_hdr_kprobe);
}
struct in6_addr src_ip = {0};
struct in6_addr dst_ip = {0};
if (eth_type == ETH_P_IP) {
struct iphdr iph = {0};
if (bpf_probe_read_kernel(&iph, sizeof(iph), data + offset) < 0) {
return false;
}
kprobe_encode_ipv4(iph.saddr, &src_ip);
kprobe_encode_ipv4(iph.daddr, &dst_ip);
} else if (eth_type == ETH_P_IPV6) {
struct ipv6hdr ip6h = {0};
if (bpf_probe_read_kernel(&ip6h, sizeof(ip6h), data + offset) < 0) {
return false;
}
kprobe_encode_ipv6(&ip6h.saddr, &src_ip);
kprobe_encode_ipv6(&ip6h.daddr, &dst_ip);
} else {
return false;
}
// Match XDP semantics: host_key is always the customer-side IP.
if (effective_direction == 1) {
__builtin_memcpy(host_key, &dst_ip, sizeof(*host_key));
} else {
__builtin_memcpy(host_key, &src_ip, sizeof(*host_key));
}
return true;
}
static __always_inline void kprobe_track_actual_bytes(
__u8 effective_direction,
struct in6_addr *host_key,
__u32 bytes
)
{
struct host_counter *counter = (struct host_counter *)bpf_map_lookup_elem(&map_traffic, host_key);
if (!counter) {
return;
}
if (effective_direction == 1) {
counter->actual_download_bytes += bytes;
} else {
counter->actual_upload_bytes += bytes;
}
}
// Determine effective direction for *egress* (dev_hard_start_xmit) packets.
// Returns 1 for download (to LAN), 2 for upload (to Internet).
static __always_inline __u8 kprobe_determine_effective_direction(
struct sk_buff *skb,
int ifindex,
int to_internet_ifindex,
int to_isp_ifindex,
__be16 internet_vlan
)
{
// On-a-stick: both directions share one ifindex, so use VLAN to infer egress.
if (to_isp_ifindex < 0) {
__be16 vlan = kprobe_current_vlan(skb);
// Out to Internet => UPLOAD (2). Out to LAN/core => DOWNLOAD (1).
return (vlan == internet_vlan) ? 2 : 1;
}
// Two-interface: infer direction from the egress interface.
// to_internet_ifindex carries uploads; to_isp_ifindex carries downloads.
return (ifindex == to_internet_ifindex) ? 2 : 1;
}
@@ -12,6 +12,8 @@
struct host_counter {
__u64 download_bytes;
__u64 upload_bytes;
__u64 actual_download_bytes;
__u64 actual_upload_bytes;
__u64 download_packets;
__u64 upload_packets;
__u64 tcp_download_packets;
+62
View File
@@ -1,8 +1,15 @@
/* SPDX-License-Identifier: GPL-2.0 */
// Minimal XDP program that passes all packets.
// Used to verify XDP functionality.
#ifndef __TARGET_ARCH_x86
#define __TARGET_ARCH_x86 1
#endif
#include <linux/bpf.h>
#include <linux/ptrace.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_core_read.h>
#include <bpf/bpf_tracing.h>
#include <linux/in6.h>
#include <linux/ip.h>
#include <linux/ipv6.h>
@@ -51,6 +58,24 @@
// If it stays at 255, we have a configuration error.
int direction = 255;
// Constant passed in during to load, containing
// either -1 (none) or the ifindex of the interface
// in each direction. Required because the kbprobe
// is global.
int to_internet_ifindex = -1;
int to_isp_ifindex = -1;
struct net_device {
int ifindex;
} __attribute__((preserve_access_index));
struct sk_buff {
struct net_device *dev;
__u16 vlan_tci;
__u32 len;
unsigned char *data;
} __attribute__((preserve_access_index));
// Also configured during loading. For "on a stick" support,
// these are mapped to the respective VLAN facing directions.
__be16 internet_vlan = 0; // Note: turn these into big-endian
@@ -511,4 +536,41 @@ int flow_reader(struct bpf_iter__bpf_map_elem *ctx)
return 0;
}
/* Runs at packet transmit */
#include "common/kprobe_helper.h"
SEC("kprobe/dev_hard_start_xmit")
int kprobe_xmit(struct pt_regs *ctx) {
// Locate the skbuff
struct sk_buff *skb = (struct sk_buff *)PT_REGS_PARM1(ctx);
// If no SKB - bail out!
if (!skb) return 0;
// Which interface called us?
int ifindex = BPF_CORE_READ(skb, dev, ifindex);
// If ifindex is not equal to EITHER to_internet or to_isp - not relevant, exit.
if (ifindex != to_internet_ifindex && ifindex != to_isp_ifindex) return 0;
__u8 effective_direction = kprobe_determine_effective_direction(
skb,
ifindex,
to_internet_ifindex,
to_isp_ifindex,
internet_vlan
);
// Keep the computation (used by subsequent xmit accounting work).
if (effective_direction == 0) return 0;
struct in6_addr host_key = {0};
if (!kprobe_build_host_key(skb, effective_direction, &host_key)) return 0;
__u32 bytes = BPF_CORE_READ(skb, len);
if (!bytes) return 0;
// Only update existing host entries; the XDP/TC path owns insertion.
kprobe_track_actual_bytes(effective_direction, &host_key, bytes);
return 0;
}
char _license[] SEC("license") = "GPL";
+26
View File
@@ -247,6 +247,32 @@ out:
return err;
}
struct bpf_link *attach_xmit_kprobe(struct lqos_kern *obj)
{
struct bpf_link *link;
long err;
link = bpf_program__attach_kprobe(obj->progs.kprobe_xmit, false, "dev_hard_start_xmit");
err = libbpf_get_error(link);
if (err) {
const char *msg = "unknown";
if (err < 0) {
msg = strerror((int)-err);
}
fprintf(stderr, "bpf_program__attach_kprobe() fails (%ld: %s)\n", err, msg);
return NULL;
}
return link;
}
void destroy_bpf_link(struct bpf_link *link)
{
if (link != NULL) {
bpf_link__destroy(link);
}
}
// Iterator code
#include <stdio.h>
#include <unistd.h>
+2
View File
@@ -10,6 +10,8 @@ extern int tc_attach_egress(int ifindex, bool verbose, struct lqos_kern *obj);
extern int tc_detach_egress(int ifindex, bool verbose, bool flush_hook, const char * ifname);
extern int tc_attach_ingress(int ifindex, bool verbose, struct lqos_kern *obj);
extern int tc_detach_ingress(int ifindex, bool verbose, bool flush_hook, const char * ifname);
extern struct bpf_link * attach_xmit_kprobe(struct lqos_kern *obj);
extern void destroy_bpf_link(struct bpf_link *link);
extern __u64 max_tracker_ips();
extern void do_not_print();
int read_tp_buffer(struct bpf_program *prog, struct bpf_map *map);
+42 -9
View File
@@ -1,5 +1,5 @@
use crate::lqos_kernel::{
InterfaceDirection, attach_xdp_and_tc_to_interface,
AttachedPrograms, InterfaceDirection, attach_xdp_and_tc_to_interface,
bpf::{self, ring_buffer_sample_fn},
unload_xdp_from_interface,
};
@@ -11,6 +11,7 @@ use parking_lot::Mutex;
/// world insists on it.
pub(crate) struct LqosKernBpfWrapper {
ptr: *mut bpf::lqos_kern,
kprobe_link: *mut bpf::bpf_link,
}
impl LqosKernBpfWrapper {
@@ -22,6 +23,14 @@ impl LqosKernBpfWrapper {
unsafe impl Sync for LqosKernBpfWrapper {}
unsafe impl Send for LqosKernBpfWrapper {}
impl Drop for LqosKernBpfWrapper {
fn drop(&mut self) {
unsafe {
bpf::destroy_bpf_link(self.kprobe_link);
}
}
}
pub(crate) static BPF_SKELETON: Lazy<Mutex<Option<LqosKernBpfWrapper>>> =
Lazy::new(|| Mutex::new(None));
@@ -58,21 +67,34 @@ impl LibreQoSKernels {
to_isp: to_isp.to_string(),
on_a_stick: false,
};
let skeleton = attach_xdp_and_tc_to_interface(
let to_internet_ifindex =
crate::lqos_kernel::interface_name_to_index(&kernel.to_internet)? as i32;
let to_isp_ifindex = crate::lqos_kernel::interface_name_to_index(&kernel.to_isp)? as i32;
let AttachedPrograms {
skeleton,
kprobe_link,
} = attach_xdp_and_tc_to_interface(
&kernel.to_internet,
to_internet_ifindex,
to_isp_ifindex,
true,
InterfaceDirection::Internet,
heimdall_event_handler,
flowbee_event_handler,
)?;
attach_xdp_and_tc_to_interface(
&kernel.to_isp,
to_internet_ifindex,
to_isp_ifindex,
false,
InterfaceDirection::IspNetwork,
heimdall_event_handler,
flowbee_event_handler,
)?;
BPF_SKELETON
.lock()
.replace(LqosKernBpfWrapper { ptr: skeleton });
BPF_SKELETON.lock().replace(LqosKernBpfWrapper {
ptr: skeleton,
kprobe_link,
});
Ok(kernel)
}
@@ -100,21 +122,32 @@ impl LibreQoSKernels {
to_isp: String::new(),
on_a_stick: true,
};
let skeleton = attach_xdp_and_tc_to_interface(
let stick_ifindex =
crate::lqos_kernel::interface_name_to_index(&kernel.to_internet)? as i32;
let AttachedPrograms {
skeleton,
kprobe_link,
} = attach_xdp_and_tc_to_interface(
&kernel.to_internet,
stick_ifindex,
-1,
true,
InterfaceDirection::OnAStick(internet_vlan, isp_vlan, stick_offset),
heimdall_event_handler,
flowbee_event_handler,
)?;
BPF_SKELETON
.lock()
.replace(LqosKernBpfWrapper { ptr: skeleton });
BPF_SKELETON.lock().replace(LqosKernBpfWrapper {
ptr: skeleton,
kprobe_link,
});
Ok(kernel)
}
}
impl Drop for LibreQoSKernels {
fn drop(&mut self) {
let skeleton = BPF_SKELETON.lock().take();
drop(skeleton);
if !self.on_a_stick {
let _ = unload_xdp_from_interface(&self.to_internet);
let _ = unload_xdp_from_interface(&self.to_isp);
+43 -2
View File
@@ -282,12 +282,38 @@ pub enum InterfaceDirection {
OnAStick(u16, u16, u32),
}
fn warn_if_tx_segmentation_offloads_enabled() {
let Ok(config) = lqos_config::load_config() else {
warn!(
"Unable to load config before attaching xmit kprobe; TX rates may be measured incorrectly if GSO/TSO offloads remain enabled."
);
return;
};
let disabled_offloads = &config.tuning.disable_offload;
let gso_disabled = disabled_offloads.iter().any(|feature| feature == "gso");
let tso_disabled = disabled_offloads.iter().any(|feature| feature == "tso");
if !gso_disabled || !tso_disabled {
warn!(
"Attaching xmit kprobe while TX segmentation offloads are not fully disabled (gso_disabled={}, tso_disabled={}); actual transmission rates may be measured incorrectly.",
gso_disabled, tso_disabled
);
}
}
pub(crate) struct AttachedPrograms {
pub(crate) skeleton: *mut lqos_kern,
pub(crate) kprobe_link: *mut bpf::bpf_link,
}
pub fn attach_xdp_and_tc_to_interface(
interface_name: &str,
to_internet_ifindex: i32,
to_isp_ifindex: i32,
attach_xmit_kprobe: bool,
direction: InterfaceDirection,
heimdall_event_handler: bpf::ring_buffer_sample_fn,
flowbee_event_handler: bpf::ring_buffer_sample_fn,
) -> Result<*mut lqos_kern> {
) -> Result<AttachedPrograms> {
check_root()?;
// If ABI changes were made to pinned maps, ensure we do not silently reuse
// incompatible versions that truncate struct values.
@@ -303,6 +329,8 @@ pub fn attach_xdp_and_tc_to_interface(
InterfaceDirection::IspNetwork => 2,
InterfaceDirection::OnAStick(..) => 3,
};
(*(*skeleton).data).to_internet_ifindex = to_internet_ifindex;
(*(*skeleton).data).to_isp_ifindex = to_isp_ifindex;
if let InterfaceDirection::OnAStick(internet, isp, stick_offset) = direction {
(*(*skeleton).bss).internet_vlan = internet.to_be();
(*(*skeleton).bss).isp_vlan = isp.to_be();
@@ -315,6 +343,16 @@ pub fn attach_xdp_and_tc_to_interface(
attach_xdp_best_available(interface_index, prog_fd, interface_name)?;
skeleton
};
let kprobe_link = if attach_xmit_kprobe {
warn_if_tx_segmentation_offloads_enabled();
let link = unsafe { bpf::attach_xmit_kprobe(skeleton) };
if link.is_null() {
return Err(Error::msg("Unable to attach kprobe to dev_hard_start_xmit"));
}
link
} else {
std::ptr::null_mut()
};
// Configure CPU Maps
let shaping_physical_cpus: Vec<u32> = match lqos_config::load_config() {
@@ -477,7 +515,10 @@ pub fn attach_xdp_and_tc_to_interface(
}
}
Ok(skeleton)
Ok(AttachedPrograms {
skeleton,
kprobe_link,
})
}
/// Safety: Direct calls to C functions
+7 -1
View File
@@ -11,6 +11,12 @@ pub struct HostCounter {
/// Upload bytes counter (keeps incrementing)
pub upload_bytes: u64,
/// Actually transmitted download bytes counter (keeps incrementing)
pub actual_download_bytes: u64,
/// Actually transmitted upload bytes counter (keeps incrementing)
pub actual_upload_bytes: u64,
/// Download packets counter (keeps incrementing)
pub download_packets: u64,
@@ -62,6 +68,6 @@ mod test {
#[test]
fn host_counter_size() {
assert_eq!(std::mem::size_of::<HostCounter>(), 112);
assert_eq!(std::mem::size_of::<HostCounter>(), 128);
}
}
+9 -11
View File
@@ -1326,20 +1326,18 @@ async fn shaper_snapshot_streaming(
request_id: u64,
reply: tokio::sync::mpsc::Sender<Message>,
) -> anyhow::Result<()> {
// Mirror node_manager ticker throughput: fetch current throughput and send compact tuple data
use lqos_bus::BusResponse;
let resp = crate::throughput_tracker::current_throughput();
if let BusResponse::CurrentThroughput {
bits_per_second,
packets_per_second,
shaped_bits_per_second,
..
} = resp
// Mirror node_manager ticker throughput: send compact tuple data.
// Note: field names are historical; values are bits per second.
{
let bits_per_second = crate::throughput_tracker::THROUGHPUT_TRACKER.actual_bits_per_second();
let shaped_bits_per_second = crate::throughput_tracker::THROUGHPUT_TRACKER
.shaped_actual_bits_per_second();
let packets_per_second = crate::throughput_tracker::THROUGHPUT_TRACKER
.packets_per_second
.as_down_up();
let message = messages::WsMessage::StreamingShaper {
request_id,
// Note: field names are historical; values are bits per second
bytes_down: bits_per_second.down,
bytes_up: bits_per_second.up,
shaped_bytes_down: shaped_bits_per_second.down,
@@ -371,6 +371,10 @@ function pushQueuingActivitySample() {
down: currentDirectionValue(latestCircuitSummary?.bytes_per_second, "down", 0) * 8,
up: currentDirectionValue(latestCircuitSummary?.bytes_per_second, "up", 0) * 8,
},
actualThroughputBps: {
down: currentDirectionValue(latestCircuitSummary?.actual_bytes_per_second, "down", 0) * 8,
up: currentDirectionValue(latestCircuitSummary?.actual_bytes_per_second, "up", 0) * 8,
},
ceilingBps: {
down: currentDirectionValue(plan, "down", 0) * 1_000_000.0,
up: currentDirectionValue(plan, "up", 0) * 1_000_000.0,
@@ -1072,16 +1076,16 @@ function applyCircuitSummary(summary) {
}
if (speedometer) {
speedometer.update(
currentDirectionValue(summary?.bytes_per_second, "down", 0) * 8,
currentDirectionValue(summary?.bytes_per_second, "up", 0) * 8,
currentDirectionValue(summary?.actual_bytes_per_second, "down", 0) * 8,
currentDirectionValue(summary?.actual_bytes_per_second, "up", 0) * 8,
currentDirectionValue(plan, "down", 0),
currentDirectionValue(plan, "up", 0)
);
}
if (totalThroughput) {
totalThroughput.update(
currentDirectionValue(summary?.bytes_per_second, "down", 0) * 8,
currentDirectionValue(summary?.bytes_per_second, "up", 0) * 8
currentDirectionValue(summary?.actual_bytes_per_second, "down", 0) * 8,
currentDirectionValue(summary?.actual_bytes_per_second, "up", 0) * 8
);
}
if (totalRetransmits) {
@@ -1106,8 +1110,8 @@ function applyDeviceLiveData(devices) {
const throughputGraph = deviceGraphs["throughputGraph_" + device.device_id];
if (throughputGraph !== undefined) {
throughputGraph.update(
toNumber(device.bytes_per_second?.down, 0) * 8,
toNumber(device.bytes_per_second?.up, 0) * 8
toNumber(device.actual_bytes_per_second?.down, 0) * 8,
toNumber(device.actual_bytes_per_second?.up, 0) * 8
);
}
@@ -1812,14 +1816,14 @@ function fillLiveDevices(devices) {
if (throughputDown !== null) {
throughputDown.innerHTML = formatThroughput(
toNumber(device.bytes_per_second?.down, 0) * 8,
toNumber(device.actual_bytes_per_second?.down, 0) * 8,
toNumber(device.plan?.down, 0)
);
}
if (throughputUp !== null) {
throughputUp.innerHTML = formatThroughput(
toNumber(device.bytes_per_second?.up, 0) * 8,
toNumber(device.actual_bytes_per_second?.up, 0) * 8,
toNumber(device.plan?.up, 0)
);
}
@@ -45,6 +45,7 @@ function throughputPaletteColor(direction, fallback) {
function getWaveformTheme(direction = "down") {
const isDark = document.documentElement.getAttribute("data-bs-theme") !== "light";
const throughputColor = throughputPaletteColor(direction, isDark ? "#4992ff" : "#d87c7c");
const actualThroughputColor = window.graphPalette?.[2] || (isDark ? "#fddd60" : "#d7ab82");
if (isDark) {
return {
@@ -58,6 +59,8 @@ function getWaveformTheme(direction = "down") {
throughputAreaTop: hexToRgba(throughputColor, 0.48),
throughputAreaMid: hexToRgba(throughputColor, 0.28),
throughputAreaBottom: "rgba(9, 20, 31, 0.03)",
actualThroughputLine: actualThroughputColor,
actualThroughputGlow: hexToRgba(actualThroughputColor, 0.26),
rttLine: "#b7a5ff",
rttGlow: "rgba(183, 165, 255, 0.32)",
rttAreaTop: "rgba(183, 165, 255, 0.18)",
@@ -83,6 +86,8 @@ function getWaveformTheme(direction = "down") {
throughputAreaTop: hexToRgba(throughputColor, 0.24),
throughputAreaMid: hexToRgba(throughputColor, 0.13),
throughputAreaBottom: "rgba(255, 255, 255, 0.02)",
actualThroughputLine: actualThroughputColor,
actualThroughputGlow: hexToRgba(actualThroughputColor, 0.16),
rttLine: "#6f63cf",
rttGlow: "rgba(111, 99, 207, 0.2)",
rttAreaTop: "rgba(111, 99, 207, 0.1)",
@@ -426,7 +431,7 @@ export class QueuingActivityWaveform extends DashboardGraph {
],
series: [
{
name: "Throughput",
name: "Enqueued",
type: "line",
xAxisIndex: 0,
yAxisIndex: 0,
@@ -448,6 +453,27 @@ export class QueuingActivityWaveform extends DashboardGraph {
},
data: [],
},
{
name: "Throughput",
type: "line",
xAxisIndex: 0,
yAxisIndex: 0,
showSymbol: false,
smooth: false,
step: "start",
lineStyle: {
width: 2.0,
type: "dashed",
color: this.colors.actualThroughputLine,
shadowBlur: 10,
shadowColor: this.colors.actualThroughputGlow,
},
itemStyle: {
opacity: 1,
},
data: [],
z: 5,
},
{
name: "Ceiling Base",
type: "line",
@@ -561,19 +587,21 @@ export class QueuingActivityWaveform extends DashboardGraph {
{ offset: 0.45, color: this.colors.throughputAreaMid },
{ offset: 1, color: this.colors.throughputAreaBottom },
]);
this.option.series[1].lineStyle.color = this.colors.ceilingInactive;
this.option.series[1].lineStyle.shadowColor = this.colors.ceilingInactiveGlow;
this.option.series[2].lineStyle.color = this.colors.ceilingActive;
this.option.series[2].lineStyle.shadowColor = this.colors.ceilingActiveGlow;
this.option.series[1].lineStyle.color = this.colors.actualThroughputLine;
this.option.series[1].lineStyle.shadowColor = this.colors.actualThroughputGlow;
this.option.series[2].lineStyle.color = this.colors.ceilingInactive;
this.option.series[2].lineStyle.shadowColor = this.colors.ceilingInactiveGlow;
this.option.series[3].lineStyle.color = this.colors.ceilingActive;
this.option.series[3].lineStyle.shadowColor = this.colors.ceilingActiveGlow;
this.option.series[4].lineStyle.color = this.colors.rttLine;
this.option.series[4].lineStyle.shadowColor = this.colors.rttGlow;
this.option.series[4].areaStyle.color = new echarts.graphic.LinearGradient(0, 0, 0, 1, [
this.option.series[4].lineStyle.color = this.colors.ceilingActive;
this.option.series[4].lineStyle.shadowColor = this.colors.ceilingActiveGlow;
this.option.series[5].lineStyle.color = this.colors.rttLine;
this.option.series[5].lineStyle.shadowColor = this.colors.rttGlow;
this.option.series[5].areaStyle.color = new echarts.graphic.LinearGradient(0, 0, 0, 1, [
{ offset: 0, color: this.colors.rttAreaTop },
{ offset: 1, color: this.colors.rttAreaBottom },
]);
this.option.series[4].markArea.data = rttMarkAreas(
this.option.series[5].markArea.data = rttMarkAreas(
this.colors,
this.rttThresholds,
this.option.yAxis[1].max || this.rttThresholds.red_ms,
@@ -625,10 +653,10 @@ export class QueuingActivityWaveform extends DashboardGraph {
setRttThresholds(rawThresholds) {
this.rttThresholds = normalizeRttThresholds(rawThresholds);
this.option.yAxis[1].max = this.rttThresholds.red_ms;
this.option.series[4].markArea.data = rttMarkAreas(this.colors, this.rttThresholds, this.rttThresholds.red_ms);
this.option.series[5].markArea.data = rttMarkAreas(this.colors, this.rttThresholds, this.rttThresholds.red_ms);
this.chart.setOption({
yAxis: [{}, { max: this.rttThresholds.red_ms }],
series: [{}, {}, {}, {}, { markArea: { data: this.option.series[4].markArea.data } }],
series: [{}, {}, {}, {}, {}, { markArea: { data: this.option.series[5].markArea.data } }],
});
this.render();
}
@@ -640,6 +668,10 @@ export class QueuingActivityWaveform extends DashboardGraph {
down: toNumber(sample.throughputBps?.down, 0),
up: toNumber(sample.throughputBps?.up, 0),
},
actualThroughputBps: {
down: toNumber(sample.actualThroughputBps?.down, 0),
up: toNumber(sample.actualThroughputBps?.up, 0),
},
ceilingBps: {
down: toNumber(sample.ceilingBps?.down, 0),
up: toNumber(sample.ceilingBps?.up, 0),
@@ -715,6 +747,13 @@ export class QueuingActivityWaveform extends DashboardGraph {
windowStart,
displayNow,
);
const actualThroughputData = directionalSeriesData(
this.samples,
this.direction,
"actualThroughputBps",
windowStart,
displayNow,
);
const rttData = directionalSeriesData(
this.samples,
this.direction,
@@ -761,9 +800,17 @@ export class QueuingActivityWaveform extends DashboardGraph {
],
series: [
{
name: "Throughput",
name: "Enqueued",
data: throughputData,
},
{
name: "Throughput",
data: actualThroughputData,
lineStyle: {
color: this.colors.actualThroughputLine,
shadowColor: this.colors.actualThroughputGlow,
},
},
{
name: "Ceiling Base",
data: ceilingSeries.base,
@@ -20,6 +20,8 @@ const TRAFFIC_FLOW_HIDE_THRESHOLD_BPS: u32 = 1_048_576;
pub struct CircuitSummaryData {
pub circuit_id: String,
pub bytes_per_second: DownUpOrder<u64>,
#[serde(default)]
pub actual_bytes_per_second: DownUpOrder<u64>,
pub rtt_current_p50_nanos: DownUpOrder<Option<u64>>,
pub tcp_retransmit_sample: DownUpOrder<TcpRetransmitSample>,
pub qoo_score: Option<f32>,
@@ -135,6 +135,10 @@
<div class="lqos-circuit-activity-legend" aria-label="Queue Dynamics Legend">
<span class="lqos-circuit-activity-legend-item">
<span class="lqos-circuit-activity-legend-dot is-throughput"></span>
<span>Enqueued</span>
</span>
<span class="lqos-circuit-activity-legend-item">
<span class="lqos-circuit-activity-legend-dot is-throughput-actual"></span>
<span>Throughput</span>
</span>
<span class="lqos-circuit-activity-legend-item">
@@ -164,7 +168,7 @@
<div class="row g-2 mt-2">
<div class="col-12 col-md-6 col-xl-3">
<div class="lqos-circuit-activity-kpi">
<div class="lqos-circuit-activity-kpi-label">Throughput</div>
<div class="lqos-circuit-activity-kpi-label">Enqueued throughput</div>
<div id="queuingActivityThroughput" class="lqos-circuit-activity-kpi-value">-</div>
</div>
</div>
@@ -556,6 +556,10 @@ html[data-bs-theme="light"] .sidebar .navbar-toggler-icon {
background: #32d3bd;
box-shadow: 0 0 10px rgba(50, 211, 189, 0.24);
}
.lqos-circuit-activity-legend-dot.is-throughput-actual {
background: #fddd60;
box-shadow: 0 0 10px rgba(253, 221, 96, 0.22);
}
.lqos-circuit-activity-legend-dot.is-qoo {
background: #8fb0ff;
box-shadow: 0 0 10px rgba(143, 176, 255, 0.22);
@@ -636,6 +640,10 @@ html[data-bs-theme="light"] .sidebar .navbar-toggler-icon {
background: #129a87;
box-shadow: 0 0 10px rgba(18, 154, 135, 0.16);
}
[data-bs-theme="light"] .lqos-circuit-activity-legend-dot.is-throughput-actual {
background: #d7ab82;
box-shadow: 0 0 10px rgba(215, 171, 130, 0.14);
}
[data-bs-theme="light"] .lqos-circuit-activity-legend-dot.is-qoo {
background: #4f79d9;
box-shadow: 0 0 10px rgba(79, 121, 217, 0.12);
@@ -90,6 +90,14 @@ fn summarize_circuit_devices(circuit: &str, devices: &[Circuit]) -> CircuitSumma
acc
});
let actual_bytes_per_second = devices
.iter()
.fold(DownUpOrder::default(), |mut acc, device| {
acc.down += device.actual_bytes_per_second.down;
acc.up += device.actual_bytes_per_second.up;
acc
});
let tcp_retransmit_sample = down_up_retransmit_sample(
DownUpOrder {
down: devices
@@ -133,6 +141,7 @@ fn summarize_circuit_devices(circuit: &str, devices: &[Circuit]) -> CircuitSumma
CircuitSummaryData {
circuit_id: circuit.to_string(),
bytes_per_second,
actual_bytes_per_second,
rtt_current_p50_nanos,
tcp_retransmit_sample,
qoo_score: qoo_score_for_circuit(circuit),
@@ -561,6 +561,7 @@ pub fn get_all_circuits() -> BusResponse {
Circuit {
ip: k.as_ip(),
bytes_per_second: v.bytes_per_second,
actual_bytes_per_second: v.actual_bytes_per_second,
median_latency: v.median_latency(),
rtt_current_p50_nanos: DownUpOrder {
down: v
@@ -679,6 +680,7 @@ pub fn get_circuit_by_id(desired_circuit_id: String) -> BusResponse {
Some(Circuit {
ip: k.as_ip(),
bytes_per_second: v.bytes_per_second,
actual_bytes_per_second: v.actual_bytes_per_second,
median_latency: v.median_latency(),
rtt_current_p50_nanos: DownUpOrder {
down: v
+7 -7
View File
@@ -310,9 +310,9 @@ fn throughput_task(
pub fn current_throughput() -> BusResponse {
let (bits_per_second, packets_per_second, shaped_bits_per_second, tcp_pps, udp_pps, icmp_pps) = {
(
THROUGHPUT_TRACKER.bits_per_second(),
THROUGHPUT_TRACKER.actual_bits_per_second(),
THROUGHPUT_TRACKER.packets_per_second(),
THROUGHPUT_TRACKER.shaped_bits_per_second(),
THROUGHPUT_TRACKER.shaped_actual_bits_per_second(),
THROUGHPUT_TRACKER.tcp_packets_per_second(),
THROUGHPUT_TRACKER.udp_packets_per_second(),
THROUGHPUT_TRACKER.icmp_packets_per_second(),
@@ -370,7 +370,7 @@ pub fn top_n(start: u32, end: u32) -> BusResponse {
.map(|(k, te)| {
(
*k,
te.bytes_per_second,
te.actual_bytes_per_second,
te.packets_per_second,
te.median_latency().unwrap_or(0.0),
te.tc_handle,
@@ -414,7 +414,7 @@ pub fn top_n_up(start: u32, end: u32) -> BusResponse {
.map(|(k, te)| {
(
*k,
te.bytes_per_second,
te.actual_bytes_per_second,
te.packets_per_second,
te.median_latency().unwrap_or(0.0),
te.tc_handle,
@@ -567,7 +567,7 @@ pub fn worst_n(start: u32, end: u32) -> BusResponse {
.map(|(k, te)| {
(
*k,
te.bytes_per_second,
te.actual_bytes_per_second,
te.packets_per_second,
te.median_latency().unwrap_or(0.0),
te.tc_handle,
@@ -612,7 +612,7 @@ pub fn worst_n_retransmits(start: u32, end: u32) -> BusResponse {
.map(|(k, te)| {
(
*k,
te.bytes_per_second,
te.actual_bytes_per_second,
te.packets_per_second,
te.median_latency().unwrap_or(0.0),
te.tc_handle,
@@ -670,7 +670,7 @@ pub fn best_n(start: u32, end: u32) -> BusResponse {
.map(|(k, te)| {
(
*k,
te.bytes_per_second,
te.actual_bytes_per_second,
te.packets_per_second,
te.median_latency().unwrap_or(0.0),
te.tc_handle,
@@ -96,7 +96,7 @@ pub(crate) fn submit_throughput_stats(
THROUGHPUT_TRACKER.icmp_packets_per_second.get_down(),
THROUGHPUT_TRACKER.icmp_packets_per_second.get_up(),
);
let bits_per_second = THROUGHPUT_TRACKER.bits_per_second();
let bits_per_second = THROUGHPUT_TRACKER.actual_bits_per_second();
// Check that the stats haven't gone wonky and don't submit obviously bad data.
if let Ok(config) = load_config() {
@@ -206,8 +206,10 @@ pub(crate) fn submit_throughput_stats(
}
// Send top-level throughput stats to LTS2
let bytes = THROUGHPUT_TRACKER.bytes_per_second.as_down_up();
let shaped_bytes = THROUGHPUT_TRACKER.shaped_bytes_per_second.as_down_up();
let bytes = THROUGHPUT_TRACKER.actual_bytes_per_second.as_down_up();
let shaped_bytes = THROUGHPUT_TRACKER
.shaped_actual_bytes_per_second
.as_down_up();
let mut min_rtt = None;
let mut max_rtt = None;
let mut median_rtt = None;
@@ -284,11 +286,11 @@ pub(crate) fn submit_throughput_stats(
.raw_data
.lock()
.iter()
.filter(|(_k, h)| h.circuit_id.is_some() && h.bytes_per_second.not_zero())
.filter(|(_k, h)| h.circuit_id.is_some() && h.actual_bytes_per_second.not_zero())
.for_each(|(_k, h)| {
let mut crazy = false;
if let Some((dl, ul)) = plan_lookup.get(&h.circuit_hash.unwrap_or(0))
&& (h.bytes_per_second.down > *dl || h.bytes_per_second.up > *ul)
&& (h.actual_bytes_per_second.down > *dl || h.actual_bytes_per_second.up > *ul)
{
crazy_values.insert(h.circuit_hash.unwrap_or(0));
crazy = true;
@@ -298,7 +300,7 @@ pub(crate) fn submit_throughput_stats(
return;
}
if let Some(c) = circuit_throughput.get_mut(&h.circuit_hash.unwrap_or(0)) {
c.bytes += h.bytes_per_second;
c.bytes += h.actual_bytes_per_second;
c.packets += h.packets_per_second;
c.tcp_packets += h.tcp_packets;
c.udp_packets += h.udp_packets;
@@ -307,7 +309,7 @@ pub(crate) fn submit_throughput_stats(
circuit_throughput.insert(
h.circuit_hash.unwrap_or(0),
CircuitThroughputTemp {
bytes: h.bytes_per_second,
bytes: h.actual_bytes_per_second,
packets: h.packets_per_second,
tcp_packets: h.tcp_packets,
udp_packets: h.udp_packets,
@@ -13,16 +13,19 @@ pub(crate) struct ThroughputEntry {
pub(crate) first_cycle: u64,
pub(crate) most_recent_cycle: u64,
pub(crate) bytes: DownUpOrder<u64>, // 0 DL, 1 UL
pub(crate) actual_bytes: DownUpOrder<u64>, // 0 DL, 1 UL
pub(crate) packets: DownUpOrder<u64>, // 0 DL, 1 UL
pub(crate) tcp_packets: DownUpOrder<u64>, // 0 DL, 1 UL
pub(crate) udp_packets: DownUpOrder<u64>, // 0 DL, 1 UL
pub(crate) icmp_packets: DownUpOrder<u64>, // 0 DL, 1 UL
pub(crate) prev_bytes: DownUpOrder<u64>, // Has to mirror
pub(crate) prev_actual_bytes: DownUpOrder<u64>,
pub(crate) prev_packets: DownUpOrder<u64>,
pub(crate) prev_tcp_packets: DownUpOrder<u64>,
pub(crate) prev_udp_packets: DownUpOrder<u64>,
pub(crate) prev_icmp_packets: DownUpOrder<u64>,
pub(crate) bytes_per_second: DownUpOrder<u64>,
pub(crate) actual_bytes_per_second: DownUpOrder<u64>,
pub(crate) packets_per_second: DownUpOrder<u64>,
pub(crate) tc_handle: TcHandle,
pub(crate) rtt_buffer: RttBuffer,
@@ -53,11 +53,13 @@ pub struct ThroughputTracker {
pub(crate) cycle: AtomicU64,
pub(crate) raw_data: Mutex<HashMap<XdpIpAddress, ThroughputEntry>>,
pub(crate) bytes_per_second: AtomicDownUp,
pub(crate) actual_bytes_per_second: AtomicDownUp,
pub(crate) packets_per_second: AtomicDownUp,
pub(crate) tcp_packets_per_second: AtomicDownUp,
pub(crate) udp_packets_per_second: AtomicDownUp,
pub(crate) icmp_packets_per_second: AtomicDownUp,
pub(crate) shaped_bytes_per_second: AtomicDownUp,
pub(crate) shaped_actual_bytes_per_second: AtomicDownUp,
pub(crate) circuit_heatmaps: Mutex<FxHashMap<i64, TemporalHeatmap>>,
pub(crate) circuit_qoq_heatmaps: Mutex<FxHashMap<i64, TemporalQoqHeatmap>>,
pub(crate) global_heatmap: Mutex<TemporalHeatmap>,
@@ -74,6 +76,7 @@ struct CircuitHeatmapAggregate {
struct ReducedHostCounters {
bytes: DownUpOrder<u64>,
actual_bytes: DownUpOrder<u64>,
packets: DownUpOrder<u64>,
tcp_packets: DownUpOrder<u64>,
udp_packets: DownUpOrder<u64>,
@@ -87,6 +90,7 @@ struct ReducedHostCounters {
impl ReducedHostCounters {
fn from_counters(counts: &[lqos_sys::HostCounter]) -> Self {
let mut bytes = DownUpOrder::zeroed();
let mut actual_bytes = DownUpOrder::zeroed();
let mut packets = DownUpOrder::zeroed();
let mut tcp_packets = DownUpOrder::zeroed();
let mut udp_packets = DownUpOrder::zeroed();
@@ -99,6 +103,8 @@ impl ReducedHostCounters {
for c in counts {
bytes.checked_add_direct(c.download_bytes, c.upload_bytes);
actual_bytes
.checked_add_direct(c.actual_download_bytes, c.actual_upload_bytes);
packets.checked_add_direct(c.download_packets, c.upload_packets);
tcp_packets.checked_add_direct(c.tcp_download_packets, c.tcp_upload_packets);
udp_packets.checked_add_direct(c.udp_download_packets, c.udp_upload_packets);
@@ -114,6 +120,7 @@ impl ReducedHostCounters {
Self {
bytes,
actual_bytes,
packets,
tcp_packets,
udp_packets,
@@ -136,11 +143,13 @@ impl ThroughputTracker {
cycle: AtomicU64::new(RETIRE_AFTER_SECONDS),
raw_data: Mutex::default(),
bytes_per_second: AtomicDownUp::zeroed(),
actual_bytes_per_second: AtomicDownUp::zeroed(),
packets_per_second: AtomicDownUp::zeroed(),
tcp_packets_per_second: AtomicDownUp::zeroed(),
udp_packets_per_second: AtomicDownUp::zeroed(),
icmp_packets_per_second: AtomicDownUp::zeroed(),
shaped_bytes_per_second: AtomicDownUp::zeroed(),
shaped_actual_bytes_per_second: AtomicDownUp::zeroed(),
circuit_heatmaps: Mutex::default(),
circuit_qoq_heatmaps: Mutex::default(),
global_heatmap: Mutex::new(TemporalHeatmap::new()),
@@ -194,8 +203,14 @@ impl ThroughputTracker {
continue;
};
let download_delta = entry.bytes.down.saturating_sub(entry.prev_bytes.down);
let upload_delta = entry.bytes.up.saturating_sub(entry.prev_bytes.up);
let download_delta = entry
.actual_bytes
.down
.saturating_sub(entry.prev_actual_bytes.down);
let upload_delta = entry
.actual_bytes
.up
.saturating_sub(entry.prev_actual_bytes.up);
total_download_bytes = total_download_bytes.saturating_add(download_delta);
total_upload_bytes = total_upload_bytes.saturating_add(upload_delta);
total_tcp_packets.down = total_tcp_packets
@@ -368,9 +383,12 @@ impl ThroughputTracker {
raw_data.iter_mut().for_each(|(_k, v)| {
if v.first_cycle < self_cycle {
v.bytes_per_second = v.bytes.checked_sub_or_zero(v.prev_bytes);
v.actual_bytes_per_second =
v.actual_bytes.checked_sub_or_zero(v.prev_actual_bytes);
v.packets_per_second = v.packets.checked_sub_or_zero(v.prev_packets);
}
v.prev_bytes = v.bytes;
v.prev_actual_bytes = v.actual_bytes;
v.prev_packets = v.packets;
v.prev_tcp_packets = v.tcp_packets;
v.prev_udp_packets = v.udp_packets;
@@ -455,6 +473,7 @@ impl ThroughputTracker {
if let Some(entry) = raw_data.get_mut(xdp_ip) {
// Zero the counter, we have to do a per-CPU sum
entry.bytes = reduced.bytes;
entry.actual_bytes = reduced.actual_bytes;
entry.packets = reduced.packets;
entry.tcp_packets = reduced.tcp_packets;
entry.udp_packets = reduced.udp_packets;
@@ -489,8 +508,14 @@ impl ThroughputTracker {
net_json_calc.add_throughput_cycle(
parents,
(
entry.bytes.down.saturating_sub(entry.prev_bytes.down),
entry.bytes.up.saturating_sub(entry.prev_bytes.up),
entry
.actual_bytes
.down
.saturating_sub(entry.prev_actual_bytes.down),
entry
.actual_bytes
.up
.saturating_sub(entry.prev_actual_bytes.up),
),
(
entry.packets.down.saturating_sub(entry.prev_packets.down),
@@ -567,10 +592,13 @@ impl ThroughputTracker {
first_cycle: self_cycle,
most_recent_cycle: 0,
bytes: reduced.bytes,
actual_bytes: reduced.actual_bytes,
packets: reduced.packets,
prev_bytes: DownUpOrder::zeroed(),
prev_actual_bytes: DownUpOrder::zeroed(),
prev_packets: DownUpOrder::zeroed(),
bytes_per_second: DownUpOrder::zeroed(),
actual_bytes_per_second: DownUpOrder::zeroed(),
packets_per_second: DownUpOrder::zeroed(),
tcp_packets: reduced.tcp_packets,
udp_packets: reduced.udp_packets,
@@ -1047,11 +1075,13 @@ impl ThroughputTracker {
pub(crate) fn update_totals(&self) {
let current_cycle = self.cycle.load(std::sync::atomic::Ordering::Relaxed);
self.bytes_per_second.set_to_zero();
self.actual_bytes_per_second.set_to_zero();
self.packets_per_second.set_to_zero();
self.tcp_packets_per_second.set_to_zero();
self.udp_packets_per_second.set_to_zero();
self.icmp_packets_per_second.set_to_zero();
self.shaped_bytes_per_second.set_to_zero();
self.shaped_actual_bytes_per_second.set_to_zero();
let raw_data = self.raw_data.lock();
raw_data
.iter()
@@ -1062,6 +1092,10 @@ impl ThroughputTracker {
(
v.bytes.down.saturating_sub(v.prev_bytes.down),
v.bytes.up.saturating_sub(v.prev_bytes.up),
v.actual_bytes
.down
.saturating_sub(v.prev_actual_bytes.down),
v.actual_bytes.up.saturating_sub(v.prev_actual_bytes.up),
v.packets.down.saturating_sub(v.prev_packets.down),
v.packets.up.saturating_sub(v.prev_packets.up),
v.tcp_packets.down.saturating_sub(v.prev_tcp_packets.down),
@@ -1077,6 +1111,8 @@ impl ThroughputTracker {
|(
bytes_down,
bytes_up,
actual_bytes_down,
actual_bytes_up,
packets_down,
packets_up,
tcp_down,
@@ -1089,6 +1125,8 @@ impl ThroughputTracker {
)| {
self.bytes_per_second
.checked_add_tuple((bytes_down, bytes_up));
self.actual_bytes_per_second
.checked_add_tuple((actual_bytes_down, actual_bytes_up));
self.packets_per_second
.checked_add_tuple((packets_down, packets_up));
self.tcp_packets_per_second
@@ -1100,6 +1138,8 @@ impl ThroughputTracker {
if shaped {
self.shaped_bytes_per_second
.checked_add_tuple((bytes_down, bytes_up));
self.shaped_actual_bytes_per_second
.checked_add_tuple((actual_bytes_down, actual_bytes_up));
}
},
);
@@ -1148,12 +1188,25 @@ impl ThroughputTracker {
self.bytes_per_second.as_down_up().to_bits_from_bytes()
}
#[allow(dead_code)]
pub(crate) fn actual_bits_per_second(&self) -> DownUpOrder<u64> {
self.actual_bytes_per_second.as_down_up().to_bits_from_bytes()
}
#[allow(dead_code)]
pub(crate) fn shaped_bits_per_second(&self) -> DownUpOrder<u64> {
self.shaped_bytes_per_second
.as_down_up()
.to_bits_from_bytes()
}
#[allow(dead_code)]
pub(crate) fn shaped_actual_bits_per_second(&self) -> DownUpOrder<u64> {
self.shaped_actual_bytes_per_second
.as_down_up()
.to_bits_from_bytes()
}
pub(crate) fn packets_per_second(&self) -> DownUpOrder<u64> {
self.packets_per_second.as_down_up()
}