This commit is contained in:
Name committed 2026-04-23 15:31:11 -06:00
1 parent 4922b90e46
commit 1ea441250d
6 files changed
+418 -13

No files matched your search

+241 -13
View File
@@ -1936,6 +1936,91 @@ impl Drop for FullReloadScope {
}
}
struct TcClassifyBypassGuard {
cleanup_on_drop: bool,
setter: fn(bool) -> anyhow::Result<()>,
}
struct BakeryFullReloadBatchResult {
result: ExecuteResult,
summary: String,
build_duration_ms: u64,
total_tc_commands: usize,
class_commands: usize,
qdisc_commands: usize,
}
impl BakeryFullReloadBatchResult {
fn metrics<'a>(&self, summary: &'a str, ok: bool) -> BakeryApplyMetrics<'a> {
BakeryApplyMetrics {
apply_type: BakeryApplyType::FullReload,
summary,
build_duration_ms: self.build_duration_ms,
apply_duration_ms: self.result.duration_ms,
total_tc_commands: self.total_tc_commands,
class_commands: self.class_commands,
qdisc_commands: self.qdisc_commands,
ok,
}
}
}
impl TcClassifyBypassGuard {
fn new() -> anyhow::Result<Self> {
Self::with_setter(lqos_sys::set_tc_classify_bypass)
}
fn with_setter(setter: fn(bool) -> anyhow::Result<()>) -> anyhow::Result<Self> {
setter(true).map_err(|error| {
error!(
"Bakery full reload failed to enable TC classify bypass via tc_classify_control: {error}"
);
error
})?;
info!("Bakery full reload enabled TC classify bypass");
Ok(Self {
cleanup_on_drop: true,
setter,
})
}
fn disable(&mut self) -> anyhow::Result<()> {
if !self.cleanup_on_drop {
return Ok(());
}
self.cleanup_on_drop = false;
(self.setter)(false).map_err(|error| {
error!(
"Bakery full reload failed to disable TC classify bypass via tc_classify_control: {error}"
);
error
})?;
info!("Bakery full reload disabled TC classify bypass");
Ok(())
}
fn keep_bypass_enabled_after_failure(&mut self, summary: &str) {
self.cleanup_on_drop = false;
error!(
"Bakery full reload keeping TC classify bypass enabled after failure: {}",
summary
);
}
}
impl Drop for TcClassifyBypassGuard {
fn drop(&mut self) {
if !self.cleanup_on_drop {
return;
}
if let Err(error) = self.disable() {
mark_reload_required(format!(
"Bakery full reload could not disable TC classify bypass after reload: {error}. Operator intervention may be required because traffic may remain unshaped."
));
}
}
}
fn telemetry_state() -> &'static RwLock<BakeryTelemetryState> {
BAKERY_TELEMETRY.get_or_init(|| RwLock::new(BakeryTelemetryState::default()))
}
@@ -10447,6 +10532,29 @@ fn full_reload(
trigger_summary,
);
let _reload_scope = FullReloadScope;
let mut tc_bypass_guard = match TcClassifyBypassGuard::new() {
Ok(guard) => guard,
Err(error) => {
let summary =
format!("Failed to enable TC classify bypass before full reload: {error}");
error!("{summary}");
mark_reload_required(format!(
"{summary}. Full reload was not attempted because TC classify could not be safely bypassed."
));
mark_bakery_action_finished(BakeryApplyMetrics {
apply_type: BakeryApplyType::FullReload,
summary: &summary,
build_duration_ms: 0,
apply_duration_ms: 0,
total_tc_commands: 0,
class_commands: 0,
qdisc_commands: 0,
ok: false,
});
*batch = None;
return;
}
};
let previous_sites = sites.clone();
let previous_circuits = circuits.clone();
let previous_live_circuits = live_circuits.clone();
@@ -10457,6 +10565,10 @@ fn full_reload(
if let Err(error) = prepare_root_mq_for_full_reload(config) {
let summary = format!("Failed to prepare root mq state before full reload: {error}");
error!("{summary}");
tc_bypass_guard.keep_bypass_enabled_after_failure(&summary);
mark_reload_required(format!(
"{summary}. TC classify bypass remains enabled to avoid stale class-handle classification until a successful reload or operator intervention."
));
mark_bakery_action_finished(BakeryApplyMetrics {
apply_type: BakeryApplyType::FullReload,
summary: &summary,
@@ -10479,6 +10591,10 @@ fn full_reload(
let summary =
format!("Failed to snapshot live qdisc handles before full reload: {error}");
error!("{summary}");
tc_bypass_guard.keep_bypass_enabled_after_failure(&summary);
mark_reload_required(format!(
"{summary}. TC classify bypass remains enabled to avoid stale class-handle classification until a successful reload or operator intervention."
));
mark_bakery_action_finished(BakeryApplyMetrics {
apply_type: BakeryApplyType::FullReload,
summary: &summary,
@@ -10502,7 +10618,7 @@ fn full_reload(
warn!("Bakery: full reload skipped MQ layout restore because layout is unknown");
}
let result = process_batch(
let batch_result = process_batch(
new_batch,
config,
&mut working_sites,
@@ -10512,7 +10628,7 @@ fn full_reload(
&live_reserved_handles,
);
if result.ok {
if batch_result.result.ok {
*sites = working_sites;
*circuits = working_circuits;
live_circuits.clear();
@@ -10538,11 +10654,28 @@ fn full_reload(
*mq_layout = previous_mq_layout;
MQ_CREATED.store(previous_mq_created, Ordering::Relaxed);
SHAPING_TREE_ACTIVE.store(previous_shaping_tree_active, Ordering::Relaxed);
let summary = "Bakery full reload failed while applying TC command batch";
tc_bypass_guard.keep_bypass_enabled_after_failure(summary);
mark_reload_required(format!(
"{summary}. TC classify bypass remains enabled to avoid stale class-handle classification until a successful reload or operator intervention."
));
}
if result.ok {
if batch_result.result.ok {
refresh_live_capacity_snapshot(config, true);
update_queue_distribution_snapshot(sites, circuits);
if let Err(error) = tc_bypass_guard.disable() {
let summary = format!(
"Bakery full reload applied TC commands but could not disable TC classify bypass: {error}. Operator intervention may be required because traffic may remain unshaped."
);
mark_reload_required(summary.clone());
mark_bakery_action_finished(batch_result.metrics(&summary, false));
} else {
mark_bakery_action_finished(batch_result.metrics(&batch_result.summary, true));
}
} else {
update_queue_distribution_snapshot(sites, circuits);
mark_bakery_action_finished(batch_result.metrics(&batch_result.summary, false));
}
update_queue_distribution_snapshot(sites, circuits);
*batch = None;
}
@@ -10554,7 +10687,7 @@ fn process_batch(
mq_layout: &MqDeviceLayout,
qdisc_handles: &mut QdiscHandleState,
extra_reserved_handles: &HashMap<String, HashSet<u16>>,
) -> ExecuteResult {
) -> BakeryFullReloadBatchResult {
info!("Bakery: Processing batch of {} commands", batch.len());
update_bakery_apply_progress(Some("Building tc command batch"), 0, 0, 0, 0);
let build_started = std::time::Instant::now();
@@ -10620,18 +10753,14 @@ fn process_batch(
);
let summary = summarize_apply_result("processing batch", &result);
maybe_emit_memory_guard_urgent(&summary);
mark_bakery_action_finished(BakeryApplyMetrics {
apply_type: BakeryApplyType::FullReload,
summary: &summary,
BakeryFullReloadBatchResult {
result,
summary,
build_duration_ms,
apply_duration_ms: result.duration_ms,
total_tc_commands,
class_commands,
qdisc_commands,
ok: result.ok,
});
result
}
}
fn apply_stormguard_overrides(
@@ -10679,8 +10808,34 @@ fn apply_stormguard_overrides(
mod tests {
use super::*;
use lqos_config::Config;
use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};
static TC_BYPASS_SETTER_CALLS: Mutex<Vec<bool>> = Mutex::new(Vec::new());
fn recording_tc_bypass_setter(enabled: bool) -> anyhow::Result<()> {
TC_BYPASS_SETTER_CALLS.lock().expect("record setter call").push(enabled);
Ok(())
}
fn failing_tc_bypass_setter(enabled: bool) -> anyhow::Result<()> {
TC_BYPASS_SETTER_CALLS.lock().expect("record setter call").push(enabled);
Err(anyhow::anyhow!("synthetic TC classify bypass setter failure"))
}
fn disable_failing_tc_bypass_setter(enabled: bool) -> anyhow::Result<()> {
TC_BYPASS_SETTER_CALLS.lock().expect("record setter call").push(enabled);
if enabled {
Ok(())
} else {
Err(anyhow::anyhow!("synthetic TC classify bypass disable failure"))
}
}
fn take_tc_bypass_setter_calls() -> Vec<bool> {
std::mem::take(&mut *TC_BYPASS_SETTER_CALLS.lock().expect("take setter calls"))
}
fn bakery_test_lock() -> &'static std::sync::Mutex<()> {
crate::test_state_lock()
}
@@ -10725,9 +10880,82 @@ mod tests {
SHAPING_TREE_ACTIVE.store(false, Ordering::Relaxed);
FIRST_COMMIT_APPLIED.store(false, Ordering::Relaxed);
FULL_RELOAD_IN_PROGRESS.store(false, Ordering::Relaxed);
take_tc_bypass_setter_calls();
install_bakery_test_config();
}
#[test]
fn tc_classify_bypass_guard_disables_on_scope_exit() {
let _guard = bakery_test_lock().lock().expect("bakery test lock");
reset_bakery_test_state();
{
let _bypass_guard =
TcClassifyBypassGuard::with_setter(recording_tc_bypass_setter)
.expect("enable bypass guard");
assert_eq!(take_tc_bypass_setter_calls(), vec![true]);
}
assert_eq!(take_tc_bypass_setter_calls(), vec![false]);
}
#[test]
fn tc_classify_bypass_guard_disables_on_early_return() {
fn early_return() -> Result<(), ()> {
let _bypass_guard =
TcClassifyBypassGuard::with_setter(recording_tc_bypass_setter)
.map_err(|_| ())?;
Err(())
}
let _guard = bakery_test_lock().lock().expect("bakery test lock");
reset_bakery_test_state();
assert_eq!(early_return(), Err(()));
assert_eq!(take_tc_bypass_setter_calls(), vec![true, false]);
}
#[test]
fn tc_classify_bypass_guard_does_not_disable_when_enable_fails() {
let _guard = bakery_test_lock().lock().expect("bakery test lock");
reset_bakery_test_state();
assert!(TcClassifyBypassGuard::with_setter(failing_tc_bypass_setter).is_err());
assert_eq!(take_tc_bypass_setter_calls(), vec![true]);
}
#[test]
fn tc_classify_bypass_guard_reports_disable_failure_once() {
let _guard = bakery_test_lock().lock().expect("bakery test lock");
reset_bakery_test_state();
{
let mut bypass_guard =
TcClassifyBypassGuard::with_setter(disable_failing_tc_bypass_setter)
.expect("enable bypass guard");
assert!(bypass_guard.disable().is_err());
}
assert_eq!(take_tc_bypass_setter_calls(), vec![true, false]);
}
#[test]
fn tc_classify_bypass_guard_keeps_bypass_enabled_after_failure() {
let _guard = bakery_test_lock().lock().expect("bakery test lock");
reset_bakery_test_state();
{
let mut bypass_guard =
TcClassifyBypassGuard::with_setter(recording_tc_bypass_setter)
.expect("enable bypass guard");
bypass_guard.keep_bypass_enabled_after_failure("test failure after TC mutation");
}
assert_eq!(take_tc_bypass_setter_calls(), vec![true]);
}
fn mk_add_circuit(hash: i64, ip_addresses: &str) -> Arc<BakeryCommands> {
Arc::new(BakeryCommands::AddCircuit {
circuit_hash: hash,
+22
View File
@@ -19,6 +19,28 @@ struct ip_hash_info {
__u64 device_id;
};
struct tc_classify_control {
__u32 bypass;
};
struct {
__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
__uint(max_entries, 1);
__type(key, __u32);
__type(value, struct tc_classify_control);
__uint(pinning, LIBBPF_PIN_BY_NAME);
} tc_classify_control SEC(".maps");
static __always_inline bool tc_classify_bypass_enabled(void) {
__u32 key = 0;
struct tc_classify_control *control = bpf_map_lookup_elem(&tc_classify_control, &key);
if (!control) {
// Fail safe: bypass classification rather than risk stale TC handles.
return true;
}
return control->bypass != 0;
}
// In on-a-stick mode, upload classes/CPUs are offset by this amount.
// This is configured by userspace at load time.
extern __u32 stick_offset;
+3
View File
@@ -266,6 +266,9 @@ int tc_iphash_to_cpu(struct __sk_buff *skb)
bpf_debug("(TC) Error: interface direction unspecified, aborting.");
return TC_ACT_OK;
}
if (tc_classify_bypass_enabled()) {
return TC_ACT_OK;
}
#ifdef VERBOSE
bpf_debug("(TC) SKB VLAN TCI: %u", skb->vlan_tci);
#endif
+2
View File
@@ -22,6 +22,7 @@ mod ip_mapping;
mod kernel_wrapper;
mod linux;
mod lqos_kernel;
mod tc_classify_control;
mod throughput;
pub use bpf_iterator::{end_flows, expire_throughput, iterate_flows};
@@ -35,4 +36,5 @@ pub use lqos_kernel::interface_name_to_index;
pub use lqos_kernel::ip_mapping_capacity;
pub use lqos_kernel::max_tracked_ips;
pub use lqos_kernel::unload_xdp_from_interface;
pub use tc_classify_control::{initialize_tc_classify_bypass, set_tc_classify_bypass};
pub use throughput::{HostCounter, throughput_for_each};
+6
View File
@@ -152,6 +152,11 @@ fn ensure_ip_mapping_maps_abi() -> Result<()> {
std::mem::size_of::<u32>() as u32,
std::mem::size_of::<u32>() as u32,
)?;
remove_incompatible_pinned_map(
"/sys/fs/bpf/tc_classify_control",
std::mem::size_of::<u32>() as u32,
std::mem::size_of::<crate::tc_classify_control::TcClassifyControl>() as u32,
)?;
remove_incompatible_pinned_map(
"/sys/fs/bpf/map_traffic",
std::mem::size_of::<XdpIpAddress>() as u32,
@@ -339,6 +344,7 @@ pub fn attach_xdp_and_tc_to_interface(
// Ensure no lingering XDP programs before loading/attaching
let _ = unload_xdp_from_interface(interface_name);
load_kernel(skeleton)?;
crate::initialize_tc_classify_bypass()?;
let prog_fd = bpf::bpf_program__fd((*skeleton).progs.xdp_prog);
attach_xdp_best_available(interface_index, prog_fd, interface_name)?;
skeleton
@@ -0,0 +1,144 @@
//! Control-map helpers for temporarily bypassing TC classification.
use crate::lqos_kernel::bpf::libbpf_num_possible_cpus;
use anyhow::{Error, Result};
use libbpf_sys::{bpf_map_update_elem, bpf_obj_get};
use nix::libc::close;
use std::{
ffi::{CString, c_void},
io,
mem::size_of,
};
use tracing::error;
const TC_CLASSIFY_CONTROL_PATH: &str = "/sys/fs/bpf/tc_classify_control";
const TC_CLASSIFY_CONTROL_KEY: u32 = 0;
const PER_CPU_VALUE_ALIGNMENT: usize = 8;
/// Rust mirror of `struct tc_classify_control` in the eBPF program.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[repr(C)]
pub struct TcClassifyControl {
/// Non-zero tells TC classify to pass packets without setting TC handles.
pub bypass: u32,
}
fn aligned_per_cpu_value_size() -> usize {
size_of::<TcClassifyControl>().div_ceil(PER_CPU_VALUE_ALIGNMENT) * PER_CPU_VALUE_ALIGNMENT
}
fn per_cpu_payload(enabled: bool, cpu_count: u32) -> Vec<u8> {
let stride = aligned_per_cpu_value_size();
let mut payload = vec![0u8; stride * cpu_count as usize];
let control = TcClassifyControl {
bypass: u32::from(enabled),
};
let bytes = control.bypass.to_ne_bytes();
for cpu in 0..cpu_count as usize {
let offset = cpu * stride;
payload[offset..offset + bytes.len()].copy_from_slice(&bytes);
}
payload
}
fn update_tc_classify_control(enabled: bool) -> Result<()> {
let path_c = CString::new(TC_CLASSIFY_CONTROL_PATH)?;
let fd = unsafe { bpf_obj_get(path_c.as_ptr()) };
if fd < 0 {
let error = io::Error::last_os_error();
let errno = error.raw_os_error().unwrap_or(0);
return Err(Error::msg(format!(
"Unable to open BPF map '{TC_CLASSIFY_CONTROL_PATH}' for TC classify bypass update (fd={fd}, errno={errno}, error={error})"
)));
}
let cpu_count = unsafe { libbpf_num_possible_cpus() };
if cpu_count <= 0 {
unsafe {
close(fd);
}
return Err(Error::msg(format!(
"Unable to determine CPU count for TC classify bypass update: libbpf_num_possible_cpus returned {cpu_count}"
)));
}
let cpu_count = u32::try_from(cpu_count).map_err(|e| {
unsafe {
close(fd);
}
Error::msg(format!(
"Unable to convert CPU count for TC classify bypass update: {e}"
))
})?;
let mut key = TC_CLASSIFY_CONTROL_KEY;
let payload = per_cpu_payload(enabled, cpu_count);
let err = unsafe {
bpf_map_update_elem(
fd,
&mut key as *mut u32 as *mut c_void,
payload.as_ptr() as *mut c_void,
0,
)
};
let update_error = (err != 0).then(io::Error::last_os_error);
unsafe {
close(fd);
}
if err != 0 {
let error = update_error.unwrap_or_else(|| io::Error::from_raw_os_error(0));
let errno = error.raw_os_error().unwrap_or(0);
error!(
"Unable to update BPF map '{}' for TC classify bypass={} (err={}, errno={}, error={})",
TC_CLASSIFY_CONTROL_PATH, enabled, err, errno, error
);
return Err(Error::msg(format!(
"Unable to update BPF map '{TC_CLASSIFY_CONTROL_PATH}' for TC classify bypass={enabled} (err={err}, errno={errno}, error={error})"
)));
}
Ok(())
}
/// Initializes TC classify bypass to disabled.
///
/// This function has side effects: it writes all per-CPU values for the pinned
/// `tc_classify_control` BPF map.
pub fn initialize_tc_classify_bypass() -> Result<()> {
update_tc_classify_control(false)
}
/// Enables or disables TC classify bypass.
///
/// This function has side effects: it writes all per-CPU values for the pinned
/// `tc_classify_control` BPF map. When enabled, TC classify returns `TC_ACT_OK`
/// before consuming XDP metadata, flow cache, hot cache, or LPM mappings.
pub fn set_tc_classify_bypass(enabled: bool) -> Result<()> {
update_tc_classify_control(enabled)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn per_cpu_payload_uses_eight_byte_stride() {
let payload = per_cpu_payload(true, 3);
assert_eq!(payload.len(), 3 * PER_CPU_VALUE_ALIGNMENT);
for cpu in 0..3 {
let offset = cpu * PER_CPU_VALUE_ALIGNMENT;
assert_eq!(
&payload[offset..offset + size_of::<u32>()],
&1u32.to_ne_bytes()
);
assert_eq!(
&payload[offset + size_of::<u32>()..offset + PER_CPU_VALUE_ALIGNMENT],
&[0, 0, 0, 0]
);
}
}
#[test]
fn disabled_payload_clears_all_per_cpu_values() {
let payload = per_cpu_payload(false, 2);
assert_eq!(payload, vec![0u8; 2 * PER_CPU_VALUE_ALIGNMENT]);
}
}