Move 17 library crates into crates/, CLI binary into bin/fxt, consolidate 10 test crates into testing/, split config crate from deployment config files. Root directory reduced from 38+ to ~17 directories. All Cargo.toml paths and build.rs proto refs updated. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1101 lines
36 KiB
Rust
1101 lines
36 KiB
Rust
//! Resource Limit Tests
|
|
//!
|
|
//! Tests system behavior at hard resource limits:
|
|
//! - File descriptor exhaustion (ulimit)
|
|
//! - Thread pool saturation (Tokio runtime)
|
|
//! - Disk space exhaustion
|
|
//! - Network bandwidth limits
|
|
//! - TCP connection limits
|
|
//! - Memory allocation limits (OOM scenarios)
|
|
//! - CPU time limits
|
|
//! - Process limit exhaustion
|
|
|
|
use anyhow::{Context, Result};
|
|
use std::fs::{File, OpenOptions};
|
|
use std::io::Write;
|
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
|
use std::sync::Arc;
|
|
use std::time::{Duration, Instant};
|
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
|
use tokio::net::{TcpListener, TcpStream};
|
|
use tokio::task::JoinSet;
|
|
use tracing::{debug, info, warn};
|
|
|
|
/// Resource limit test metrics
|
|
#[derive(Debug, Clone)]
|
|
pub struct ResourceLimitMetrics {
|
|
/// Test name
|
|
pub test_name: String,
|
|
/// Total operations attempted
|
|
pub total_operations: u64,
|
|
/// Successful operations before limit
|
|
pub successful_operations: u64,
|
|
/// Failed operations at/after limit
|
|
pub failed_operations: u64,
|
|
/// Time until limit reached
|
|
pub time_to_limit: Duration,
|
|
/// System recovered after limit
|
|
pub recovery_successful: bool,
|
|
/// Recovery time (if applicable)
|
|
pub recovery_time: Duration,
|
|
/// System remained stable (no crash)
|
|
pub system_stable: bool,
|
|
/// Graceful degradation observed
|
|
pub graceful_degradation: bool,
|
|
/// Peak resource usage
|
|
pub peak_resource_usage: u64,
|
|
/// Test duration
|
|
pub duration: Duration,
|
|
}
|
|
|
|
impl ResourceLimitMetrics {
|
|
pub fn new(test_name: &str) -> Self {
|
|
Self {
|
|
test_name: test_name.to_string(),
|
|
total_operations: 0,
|
|
successful_operations: 0,
|
|
failed_operations: 0,
|
|
time_to_limit: Duration::ZERO,
|
|
recovery_successful: false,
|
|
recovery_time: Duration::ZERO,
|
|
system_stable: true,
|
|
graceful_degradation: false,
|
|
peak_resource_usage: 0,
|
|
duration: Duration::ZERO,
|
|
}
|
|
}
|
|
|
|
/// Calculate success rate before limit
|
|
pub fn success_rate(&self) -> f64 {
|
|
if self.total_operations == 0 {
|
|
return 0.0;
|
|
}
|
|
(self.successful_operations as f64 / self.total_operations as f64) * 100.0
|
|
}
|
|
|
|
/// Calculate failure rate at/after limit
|
|
pub fn failure_rate(&self) -> f64 {
|
|
if self.total_operations == 0 {
|
|
return 0.0;
|
|
}
|
|
(self.failed_operations as f64 / self.total_operations as f64) * 100.0
|
|
}
|
|
}
|
|
|
|
/// File descriptor exhaustion test
|
|
pub struct FileDescriptorExhaustionTest {
|
|
/// Target number of open files
|
|
target_files: usize,
|
|
/// Test duration
|
|
duration: Duration,
|
|
/// Metrics
|
|
metrics: Arc<parking_lot::Mutex<ResourceLimitMetrics>>,
|
|
/// Operation counter
|
|
operation_counter: Arc<AtomicU64>,
|
|
/// Success counter
|
|
success_counter: Arc<AtomicU64>,
|
|
/// Limit reached flag
|
|
limit_reached: Arc<AtomicBool>,
|
|
/// Limit reached time
|
|
limit_time: Arc<parking_lot::Mutex<Option<Instant>>>,
|
|
}
|
|
|
|
impl FileDescriptorExhaustionTest {
|
|
pub fn new(target_files: usize, duration: Duration) -> Self {
|
|
Self {
|
|
target_files,
|
|
duration,
|
|
metrics: Arc::new(parking_lot::Mutex::new(ResourceLimitMetrics::new(
|
|
"File Descriptor Exhaustion",
|
|
))),
|
|
operation_counter: Arc::new(AtomicU64::new(0)),
|
|
success_counter: Arc::new(AtomicU64::new(0)),
|
|
limit_reached: Arc::new(AtomicBool::new(false)),
|
|
limit_time: Arc::new(parking_lot::Mutex::new(None)),
|
|
}
|
|
}
|
|
|
|
/// Run file descriptor exhaustion test
|
|
///
|
|
/// # Errors
|
|
/// Returns error if the operation fails
|
|
pub async fn run(&self) -> Result<ResourceLimitMetrics> {
|
|
info!(
|
|
"Running file descriptor exhaustion test: {} target files",
|
|
self.target_files
|
|
);
|
|
|
|
let start = Instant::now();
|
|
let temp_dir = tempfile::tempdir().context("Failed to create temp dir")?;
|
|
let mut open_files: Vec<File> = Vec::new();
|
|
|
|
// Phase 1: Open files until we hit the limit
|
|
for i in 0..self.target_files {
|
|
self.operation_counter.fetch_add(1, Ordering::Relaxed);
|
|
|
|
let file_path = temp_dir.path().join(format!("test_file_{}.txt", i));
|
|
match OpenOptions::new()
|
|
.create(true)
|
|
.truncate(true)
|
|
.write(true)
|
|
.read(true)
|
|
.open(&file_path)
|
|
{
|
|
Ok(mut file) => {
|
|
// Write some data to ensure file is fully opened
|
|
if file.write_all(b"test data").is_ok() {
|
|
open_files.push(file);
|
|
self.success_counter.fetch_add(1, Ordering::Relaxed);
|
|
|
|
if i % 100 == 0 {
|
|
debug!("Opened {} files", i + 1);
|
|
}
|
|
}
|
|
},
|
|
Err(e) => {
|
|
if !self.limit_reached.swap(true, Ordering::Relaxed) {
|
|
let mut time = self.limit_time.lock();
|
|
if time.is_none() {
|
|
*time = Some(Instant::now());
|
|
warn!(
|
|
"File descriptor limit reached at {} files: {}",
|
|
open_files.len(),
|
|
e
|
|
);
|
|
}
|
|
}
|
|
break;
|
|
},
|
|
}
|
|
|
|
if start.elapsed() >= self.duration {
|
|
break;
|
|
}
|
|
}
|
|
|
|
let peak_files = open_files.len();
|
|
info!("Peak open files: {}", peak_files);
|
|
|
|
// Phase 2: Close some files and verify recovery
|
|
let files_to_close = open_files.len() / 2;
|
|
open_files.truncate(files_to_close);
|
|
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
|
|
// Phase 3: Try opening more files to verify recovery
|
|
let recovery_start = Instant::now();
|
|
let mut recovery_successful = false;
|
|
|
|
for i in 0..10 {
|
|
let file_path = temp_dir.path().join(format!("recovery_file_{}.txt", i));
|
|
if OpenOptions::new()
|
|
.create(true)
|
|
.truncate(true)
|
|
.write(true)
|
|
.open(&file_path)
|
|
.is_ok()
|
|
{
|
|
recovery_successful = true;
|
|
break;
|
|
}
|
|
}
|
|
|
|
let recovery_time = recovery_start.elapsed();
|
|
|
|
// Finalize metrics
|
|
let mut metrics = self.metrics.lock();
|
|
metrics.duration = start.elapsed();
|
|
metrics.total_operations = self.operation_counter.load(Ordering::Relaxed);
|
|
metrics.successful_operations = self.success_counter.load(Ordering::Relaxed);
|
|
metrics.failed_operations = metrics.total_operations - metrics.successful_operations;
|
|
metrics.peak_resource_usage = peak_files as u64;
|
|
metrics.recovery_successful = recovery_successful;
|
|
metrics.recovery_time = recovery_time;
|
|
metrics.graceful_degradation = self.limit_reached.load(Ordering::Relaxed);
|
|
|
|
if let Some(limit_instant) = *self.limit_time.lock() {
|
|
metrics.time_to_limit = limit_instant.duration_since(start);
|
|
}
|
|
|
|
info!("File descriptor exhaustion test complete: {:?}", metrics);
|
|
|
|
Ok(metrics.clone())
|
|
}
|
|
}
|
|
|
|
/// Thread pool exhaustion test (Tokio runtime)
|
|
pub struct ThreadPoolExhaustionTest {
|
|
/// Number of tasks to spawn
|
|
target_tasks: usize,
|
|
/// Test duration
|
|
duration: Duration,
|
|
/// Metrics
|
|
metrics: Arc<parking_lot::Mutex<ResourceLimitMetrics>>,
|
|
}
|
|
|
|
impl ThreadPoolExhaustionTest {
|
|
pub fn new(target_tasks: usize, duration: Duration) -> Self {
|
|
Self {
|
|
target_tasks,
|
|
duration,
|
|
metrics: Arc::new(parking_lot::Mutex::new(ResourceLimitMetrics::new(
|
|
"Thread Pool Exhaustion",
|
|
))),
|
|
}
|
|
}
|
|
|
|
/// Run thread pool exhaustion test
|
|
///
|
|
/// # Errors
|
|
/// Returns error if the operation fails
|
|
pub async fn run(&self) -> Result<ResourceLimitMetrics> {
|
|
info!(
|
|
"Running thread pool exhaustion test: {} tasks",
|
|
self.target_tasks
|
|
);
|
|
|
|
let start = Instant::now();
|
|
let mut join_set = JoinSet::new();
|
|
let tasks_spawned = Arc::new(AtomicU64::new(0));
|
|
let tasks_completed = Arc::new(AtomicU64::new(0));
|
|
|
|
// Spawn many blocking tasks
|
|
for i in 0..self.target_tasks {
|
|
let tasks_spawned = Arc::clone(&tasks_spawned);
|
|
let tasks_completed = Arc::clone(&tasks_completed);
|
|
let task_duration = self.duration;
|
|
|
|
// Use spawn_blocking to saturate the blocking thread pool
|
|
let handle = tokio::task::spawn_blocking(move || {
|
|
tasks_spawned.fetch_add(1, Ordering::Relaxed);
|
|
|
|
let task_start = Instant::now();
|
|
// Simulate blocking work
|
|
while task_start.elapsed() < task_duration.min(Duration::from_secs(1)) {
|
|
// CPU-intensive work
|
|
let mut x = i as f64;
|
|
for _ in 0..10_000 {
|
|
x = (x * 1.1).sin().cos();
|
|
}
|
|
std::thread::sleep(Duration::from_micros(100));
|
|
}
|
|
|
|
tasks_completed.fetch_add(1, Ordering::Relaxed);
|
|
});
|
|
|
|
join_set.spawn(handle);
|
|
|
|
if i % 100 == 0 {
|
|
debug!("Spawned {} tasks", i + 1);
|
|
}
|
|
|
|
// Brief yield to allow scheduling
|
|
if i % 10 == 0 {
|
|
tokio::task::yield_now().await;
|
|
}
|
|
|
|
if start.elapsed() >= self.duration {
|
|
break;
|
|
}
|
|
}
|
|
|
|
info!("Waiting for tasks to complete...");
|
|
|
|
// Wait for all tasks with timeout
|
|
let completion_timeout = Duration::from_secs(30);
|
|
let completion_start = Instant::now();
|
|
|
|
while join_set.join_next().await.is_some() {
|
|
if completion_start.elapsed() >= completion_timeout {
|
|
warn!("Task completion timeout reached, aborting remaining tasks");
|
|
join_set.abort_all();
|
|
break;
|
|
}
|
|
}
|
|
|
|
// Finalize metrics
|
|
let mut metrics = self.metrics.lock();
|
|
metrics.duration = start.elapsed();
|
|
metrics.total_operations = tasks_spawned.load(Ordering::Relaxed);
|
|
metrics.successful_operations = tasks_completed.load(Ordering::Relaxed);
|
|
metrics.failed_operations = metrics.total_operations - metrics.successful_operations;
|
|
metrics.graceful_degradation = true;
|
|
|
|
info!("Thread pool exhaustion test complete: {:?}", metrics);
|
|
|
|
Ok(metrics.clone())
|
|
}
|
|
}
|
|
|
|
/// TCP connection limit test
|
|
pub struct TcpConnectionLimitTest {
|
|
/// Target number of connections
|
|
target_connections: usize,
|
|
/// Test duration
|
|
duration: Duration,
|
|
/// Server port
|
|
port: u16,
|
|
/// Metrics
|
|
metrics: Arc<parking_lot::Mutex<ResourceLimitMetrics>>,
|
|
}
|
|
|
|
impl TcpConnectionLimitTest {
|
|
pub fn new(target_connections: usize, duration: Duration, port: u16) -> Self {
|
|
Self {
|
|
target_connections,
|
|
duration,
|
|
port,
|
|
metrics: Arc::new(parking_lot::Mutex::new(ResourceLimitMetrics::new(
|
|
"TCP Connection Limit",
|
|
))),
|
|
}
|
|
}
|
|
|
|
/// Run TCP connection limit test
|
|
///
|
|
/// # Errors
|
|
/// Returns error if the operation fails
|
|
pub async fn run(&self) -> Result<ResourceLimitMetrics> {
|
|
info!(
|
|
"Running TCP connection limit test: {} connections on port {}",
|
|
self.target_connections, self.port
|
|
);
|
|
|
|
let start = Instant::now();
|
|
|
|
// Start TCP server
|
|
let listener = TcpListener::bind(format!("127.0.0.1:{}", self.port))
|
|
.await
|
|
.context("Failed to bind TCP listener")?;
|
|
|
|
let server_handle = tokio::spawn(async move { Self::run_server(listener).await });
|
|
|
|
// Give server time to start
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
|
|
// Open many connections
|
|
let mut connections: Vec<TcpStream> = Vec::new();
|
|
let mut successful = 0;
|
|
let mut failed = 0;
|
|
|
|
for i in 0..self.target_connections {
|
|
match tokio::time::timeout(
|
|
Duration::from_secs(1),
|
|
TcpStream::connect(format!("127.0.0.1:{}", self.port)),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(stream)) => {
|
|
connections.push(stream);
|
|
successful += 1;
|
|
|
|
if i % 100 == 0 {
|
|
debug!("Opened {} connections", i + 1);
|
|
}
|
|
},
|
|
Ok(Err(e)) => {
|
|
warn!("Failed to connect: {}", e);
|
|
failed += 1;
|
|
},
|
|
Err(_) => {
|
|
warn!("Connection timeout");
|
|
failed += 1;
|
|
},
|
|
}
|
|
|
|
if start.elapsed() >= self.duration {
|
|
break;
|
|
}
|
|
}
|
|
|
|
let peak_connections = connections.len();
|
|
info!("Peak connections: {}", peak_connections);
|
|
|
|
// Close half the connections
|
|
connections.truncate(peak_connections / 2);
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
|
|
// Test recovery
|
|
let recovery_start = Instant::now();
|
|
let recovery_result = tokio::time::timeout(
|
|
Duration::from_secs(1),
|
|
TcpStream::connect(format!("127.0.0.1:{}", self.port)),
|
|
)
|
|
.await;
|
|
let recovery_successful = recovery_result.is_ok() && recovery_result.unwrap().is_ok();
|
|
let recovery_time = recovery_start.elapsed();
|
|
|
|
// Cleanup
|
|
drop(connections);
|
|
server_handle.abort();
|
|
|
|
// Finalize metrics
|
|
let mut metrics = self.metrics.lock();
|
|
metrics.duration = start.elapsed();
|
|
metrics.total_operations = (successful + failed) as u64;
|
|
metrics.successful_operations = successful as u64;
|
|
metrics.failed_operations = failed as u64;
|
|
metrics.peak_resource_usage = peak_connections as u64;
|
|
metrics.recovery_successful = recovery_successful;
|
|
metrics.recovery_time = recovery_time;
|
|
metrics.graceful_degradation = failed > 0;
|
|
|
|
info!("TCP connection limit test complete: {:?}", metrics);
|
|
|
|
Ok(metrics.clone())
|
|
}
|
|
|
|
async fn run_server(listener: TcpListener) -> Result<()> {
|
|
loop {
|
|
match listener.accept().await {
|
|
Ok((mut socket, _addr)) => {
|
|
tokio::spawn(async move {
|
|
let mut buf = [0u8; 1024];
|
|
loop {
|
|
match socket.read(&mut buf).await {
|
|
Ok(0) => break, // Connection closed
|
|
Ok(n) => {
|
|
// Echo back
|
|
if socket.write_all(&buf[..n]).await.is_err() {
|
|
break;
|
|
}
|
|
},
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
});
|
|
},
|
|
Err(_e) => {
|
|
// Server error, continue accepting
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
},
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Disk space exhaustion test
|
|
pub struct DiskSpaceExhaustionTest {
|
|
/// Target disk usage (MB)
|
|
target_disk_mb: usize,
|
|
/// Test duration
|
|
duration: Duration,
|
|
/// Metrics
|
|
metrics: Arc<parking_lot::Mutex<ResourceLimitMetrics>>,
|
|
}
|
|
|
|
impl DiskSpaceExhaustionTest {
|
|
pub fn new(target_disk_mb: usize, duration: Duration) -> Self {
|
|
Self {
|
|
target_disk_mb,
|
|
duration,
|
|
metrics: Arc::new(parking_lot::Mutex::new(ResourceLimitMetrics::new(
|
|
"Disk Space Exhaustion",
|
|
))),
|
|
}
|
|
}
|
|
|
|
/// Run disk space exhaustion test
|
|
///
|
|
/// # Errors
|
|
/// Returns error if the operation fails
|
|
pub async fn run(&self) -> Result<ResourceLimitMetrics> {
|
|
info!(
|
|
"Running disk space exhaustion test: {} MB target",
|
|
self.target_disk_mb
|
|
);
|
|
|
|
let start = Instant::now();
|
|
let temp_dir = tempfile::tempdir().context("Failed to create temp dir")?;
|
|
|
|
let chunk_size_mb = 10;
|
|
let num_chunks = self.target_disk_mb / chunk_size_mb;
|
|
let mut files_created = 0;
|
|
let mut bytes_written: u64 = 0;
|
|
|
|
// Write large files to disk
|
|
for i in 0..num_chunks {
|
|
if start.elapsed() >= self.duration {
|
|
break;
|
|
}
|
|
|
|
let file_path = temp_dir.path().join(format!("disk_test_{}.dat", i));
|
|
|
|
match std::fs::OpenOptions::new()
|
|
.create(true)
|
|
.truncate(true)
|
|
.write(true)
|
|
.open(&file_path)
|
|
{
|
|
Ok(mut file) => {
|
|
// Write chunk_size_mb of data
|
|
let data = vec![0u8; chunk_size_mb * 1_048_576];
|
|
match file.write_all(&data) {
|
|
Ok(()) => {
|
|
bytes_written += data.len() as u64;
|
|
files_created += 1;
|
|
|
|
if i % 10 == 0 {
|
|
debug!("Written {} MB", (i + 1) * chunk_size_mb);
|
|
}
|
|
},
|
|
Err(e) => {
|
|
warn!("Failed to write file: {}", e);
|
|
break;
|
|
},
|
|
}
|
|
},
|
|
Err(e) => {
|
|
warn!("Failed to create file: {}", e);
|
|
break;
|
|
},
|
|
}
|
|
|
|
// Yield to prevent blocking
|
|
tokio::task::yield_now().await;
|
|
}
|
|
|
|
let disk_mb_used = bytes_written / 1_048_576;
|
|
info!("Disk space used: {} MB", disk_mb_used);
|
|
|
|
// Finalize metrics
|
|
let mut metrics = self.metrics.lock();
|
|
metrics.duration = start.elapsed();
|
|
metrics.total_operations = num_chunks as u64;
|
|
metrics.successful_operations = files_created;
|
|
metrics.failed_operations = num_chunks as u64 - files_created;
|
|
metrics.peak_resource_usage = disk_mb_used;
|
|
metrics.graceful_degradation = true;
|
|
|
|
info!("Disk space exhaustion test complete: {:?}", metrics);
|
|
|
|
Ok(metrics.clone())
|
|
}
|
|
}
|
|
|
|
/// Memory allocation limit test (approach OOM)
|
|
pub struct MemoryAllocationLimitTest {
|
|
/// Target memory allocation (MB)
|
|
target_memory_mb: usize,
|
|
/// Allocation chunk size (MB)
|
|
chunk_size_mb: usize,
|
|
/// Test duration
|
|
duration: Duration,
|
|
/// Metrics
|
|
metrics: Arc<parking_lot::Mutex<ResourceLimitMetrics>>,
|
|
}
|
|
|
|
impl MemoryAllocationLimitTest {
|
|
pub fn new(target_memory_mb: usize, chunk_size_mb: usize, duration: Duration) -> Self {
|
|
Self {
|
|
target_memory_mb,
|
|
chunk_size_mb,
|
|
duration,
|
|
metrics: Arc::new(parking_lot::Mutex::new(ResourceLimitMetrics::new(
|
|
"Memory Allocation Limit",
|
|
))),
|
|
}
|
|
}
|
|
|
|
/// Run memory allocation limit test
|
|
///
|
|
/// # Errors
|
|
/// Returns error if the operation fails
|
|
pub async fn run(&self) -> Result<ResourceLimitMetrics> {
|
|
info!(
|
|
"Running memory allocation limit test: {} MB target (chunks of {} MB)",
|
|
self.target_memory_mb, self.chunk_size_mb
|
|
);
|
|
|
|
let start = Instant::now();
|
|
let mut allocations: Vec<Vec<u8>> = Vec::new();
|
|
let mut successful_allocations = 0;
|
|
let mut failed_allocations = 0;
|
|
|
|
let num_chunks = self.target_memory_mb / self.chunk_size_mb;
|
|
|
|
for i in 0..num_chunks {
|
|
if start.elapsed() >= self.duration {
|
|
break;
|
|
}
|
|
|
|
// Try to allocate chunk (using try_reserve to handle allocation failures gracefully)
|
|
let mut chunk = Vec::new();
|
|
let result = chunk.try_reserve_exact(self.chunk_size_mb * 1_048_576);
|
|
|
|
match result {
|
|
Ok(()) => {
|
|
// Fill the vector
|
|
chunk.resize(self.chunk_size_mb * 1_048_576, 0u8);
|
|
allocations.push(chunk);
|
|
successful_allocations += 1;
|
|
|
|
if i % 10 == 0 {
|
|
debug!("Allocated {} MB", (i + 1) * self.chunk_size_mb);
|
|
}
|
|
},
|
|
Err(_) => {
|
|
warn!(
|
|
"Memory allocation failed at {} MB",
|
|
allocations.len() * self.chunk_size_mb
|
|
);
|
|
failed_allocations += 1;
|
|
break;
|
|
},
|
|
}
|
|
|
|
// Brief yield
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
|
|
let peak_memory_mb = allocations.len() * self.chunk_size_mb;
|
|
info!("Peak memory allocated: {} MB", peak_memory_mb);
|
|
|
|
// Test recovery: release half the memory
|
|
allocations.truncate(allocations.len() / 2);
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
|
|
// Try allocating again
|
|
let recovery_start = Instant::now();
|
|
let mut recovery_vec: Vec<u8> = Vec::new();
|
|
let recovery_result = recovery_vec.try_reserve_exact(self.chunk_size_mb * 1_048_576);
|
|
let recovery_successful = recovery_result.is_ok();
|
|
let recovery_time = recovery_start.elapsed();
|
|
|
|
// Finalize metrics
|
|
let mut metrics = self.metrics.lock();
|
|
metrics.duration = start.elapsed();
|
|
metrics.total_operations = (successful_allocations + failed_allocations) as u64;
|
|
metrics.successful_operations = successful_allocations as u64;
|
|
metrics.failed_operations = failed_allocations as u64;
|
|
metrics.peak_resource_usage = peak_memory_mb as u64;
|
|
metrics.recovery_successful = recovery_successful;
|
|
metrics.recovery_time = recovery_time;
|
|
metrics.graceful_degradation = failed_allocations > 0;
|
|
|
|
info!("Memory allocation limit test complete: {:?}", metrics);
|
|
|
|
Ok(metrics.clone())
|
|
}
|
|
}
|
|
|
|
/// Network bandwidth limit test
|
|
pub struct NetworkBandwidthLimitTest {
|
|
/// Target bandwidth (MB/s)
|
|
target_bandwidth_mbps: usize,
|
|
/// Test duration
|
|
duration: Duration,
|
|
/// Server port
|
|
port: u16,
|
|
/// Metrics
|
|
metrics: Arc<parking_lot::Mutex<ResourceLimitMetrics>>,
|
|
}
|
|
|
|
impl NetworkBandwidthLimitTest {
|
|
pub fn new(target_bandwidth_mbps: usize, duration: Duration, port: u16) -> Self {
|
|
Self {
|
|
target_bandwidth_mbps,
|
|
duration,
|
|
port,
|
|
metrics: Arc::new(parking_lot::Mutex::new(ResourceLimitMetrics::new(
|
|
"Network Bandwidth Limit",
|
|
))),
|
|
}
|
|
}
|
|
|
|
/// Run network bandwidth limit test
|
|
///
|
|
/// # Errors
|
|
/// Returns error if the operation fails
|
|
pub async fn run(&self) -> Result<ResourceLimitMetrics> {
|
|
info!(
|
|
"Running network bandwidth limit test: {} MB/s target on port {}",
|
|
self.target_bandwidth_mbps, self.port
|
|
);
|
|
|
|
let start = Instant::now();
|
|
let bytes_sent = Arc::new(AtomicU64::new(0));
|
|
let bytes_received = Arc::new(AtomicU64::new(0));
|
|
|
|
// Start TCP server
|
|
let listener = TcpListener::bind(format!("127.0.0.1:{}", self.port))
|
|
.await
|
|
.context("Failed to bind TCP listener")?;
|
|
|
|
let bytes_received_clone = Arc::clone(&bytes_received);
|
|
let server_handle = tokio::spawn(async move {
|
|
Self::run_bandwidth_server(listener, bytes_received_clone).await
|
|
});
|
|
|
|
// Give server time to start
|
|
tokio::time::sleep(Duration::from_millis(100)).await;
|
|
|
|
// Start multiple clients sending data
|
|
let num_clients = 5;
|
|
let mut client_handles = Vec::new();
|
|
|
|
for _ in 0..num_clients {
|
|
let port = self.port;
|
|
let duration = self.duration;
|
|
let bytes_sent = Arc::clone(&bytes_sent);
|
|
|
|
let handle = tokio::spawn(async move {
|
|
Self::run_bandwidth_client(port, duration, bytes_sent).await
|
|
});
|
|
|
|
client_handles.push(handle);
|
|
}
|
|
|
|
// Wait for clients
|
|
for handle in client_handles {
|
|
let _ = handle.await;
|
|
}
|
|
|
|
// Cleanup
|
|
server_handle.abort();
|
|
|
|
// Calculate bandwidth
|
|
let total_bytes_sent = bytes_sent.load(Ordering::Relaxed);
|
|
let total_bytes_received = bytes_received.load(Ordering::Relaxed);
|
|
let duration_secs = start.elapsed().as_secs_f64();
|
|
let bandwidth_mbps = (total_bytes_sent as f64 / 1_048_576.0) / duration_secs;
|
|
|
|
info!(
|
|
"Bandwidth test: sent {} MB, received {} MB, {:.2} MB/s",
|
|
total_bytes_sent / 1_048_576,
|
|
total_bytes_received / 1_048_576,
|
|
bandwidth_mbps
|
|
);
|
|
|
|
// Finalize metrics
|
|
let mut metrics = self.metrics.lock();
|
|
metrics.duration = start.elapsed();
|
|
metrics.successful_operations = total_bytes_sent / 1_048_576;
|
|
metrics.peak_resource_usage = bandwidth_mbps as u64;
|
|
metrics.graceful_degradation = true;
|
|
|
|
info!("Network bandwidth limit test complete: {:?}", metrics);
|
|
|
|
Ok(metrics.clone())
|
|
}
|
|
|
|
async fn run_bandwidth_server(
|
|
listener: TcpListener,
|
|
bytes_received: Arc<AtomicU64>,
|
|
) -> Result<()> {
|
|
loop {
|
|
match listener.accept().await {
|
|
Ok((mut socket, _addr)) => {
|
|
let bytes_received = Arc::clone(&bytes_received);
|
|
tokio::spawn(async move {
|
|
let mut buf = vec![0u8; 65536];
|
|
loop {
|
|
match socket.read(&mut buf).await {
|
|
Ok(0) => break,
|
|
Ok(n) => {
|
|
bytes_received.fetch_add(n as u64, Ordering::Relaxed);
|
|
},
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
});
|
|
},
|
|
Err(_) => {
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn run_bandwidth_client(
|
|
port: u16,
|
|
duration: Duration,
|
|
bytes_sent: Arc<AtomicU64>,
|
|
) -> Result<()> {
|
|
let mut stream = TcpStream::connect(format!("127.0.0.1:{}", port)).await?;
|
|
let start = Instant::now();
|
|
let data = vec![0u8; 65536]; // 64 KB chunks
|
|
|
|
while start.elapsed() < duration {
|
|
if stream.write_all(&data).await.is_ok() {
|
|
bytes_sent.fetch_add(data.len() as u64, Ordering::Relaxed);
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn test_file_descriptor_exhaustion() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
let test = FileDescriptorExhaustionTest::new(1000, Duration::from_secs(10));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.successful_operations > 0,
|
|
"Should have opened some files"
|
|
);
|
|
assert!(metrics.system_stable, "System should remain stable");
|
|
|
|
info!(
|
|
"Opened {} files before limit",
|
|
metrics.successful_operations
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_thread_pool_exhaustion() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
let test = ThreadPoolExhaustionTest::new(500, Duration::from_secs(5));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(metrics.total_operations > 0, "Should have spawned tasks");
|
|
assert!(
|
|
metrics.successful_operations > 0,
|
|
"Some tasks should complete"
|
|
);
|
|
assert!(metrics.system_stable, "System should remain stable");
|
|
|
|
info!(
|
|
"Spawned {} tasks, {} completed",
|
|
metrics.total_operations, metrics.successful_operations
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_tcp_connection_limit() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
let test = TcpConnectionLimitTest::new(500, Duration::from_secs(10), 19000);
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.successful_operations > 0,
|
|
"Should have opened connections"
|
|
);
|
|
assert!(metrics.system_stable, "System should remain stable");
|
|
|
|
info!("Opened {} TCP connections", metrics.successful_operations);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_disk_space_exhaustion() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
// Only write 100 MB to avoid filling disk
|
|
let test = DiskSpaceExhaustionTest::new(100, Duration::from_secs(10));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.successful_operations > 0,
|
|
"Should have written files"
|
|
);
|
|
assert!(metrics.system_stable, "System should remain stable");
|
|
|
|
info!("Used {} MB disk space", metrics.peak_resource_usage);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_memory_allocation_limit() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
// Allocate up to 500 MB in 10 MB chunks
|
|
let test = MemoryAllocationLimitTest::new(500, 10, Duration::from_secs(15));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.successful_operations > 0,
|
|
"Should have allocated memory"
|
|
);
|
|
assert!(metrics.system_stable, "System should remain stable");
|
|
|
|
info!("Allocated {} MB peak memory", metrics.peak_resource_usage);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_network_bandwidth_limit() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
let test = NetworkBandwidthLimitTest::new(100, Duration::from_secs(5), 19001);
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(metrics.system_stable, "System should remain stable");
|
|
|
|
info!("Achieved {:.2} MB/s bandwidth", metrics.peak_resource_usage);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_recovery_after_file_descriptor_limit() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
let test = FileDescriptorExhaustionTest::new(1000, Duration::from_secs(10));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.recovery_successful,
|
|
"Should recover after closing files"
|
|
);
|
|
assert!(
|
|
metrics.recovery_time < Duration::from_secs(1),
|
|
"Recovery should be fast"
|
|
);
|
|
|
|
info!("Recovery time: {:?}", metrics.recovery_time);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_recovery_after_tcp_connection_limit() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
let test = TcpConnectionLimitTest::new(500, Duration::from_secs(10), 19002);
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.recovery_successful,
|
|
"Should recover after closing connections"
|
|
);
|
|
|
|
info!("TCP connection recovery successful");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_recovery_after_memory_pressure() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
let test = MemoryAllocationLimitTest::new(500, 10, Duration::from_secs(15));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.recovery_successful,
|
|
"Should recover after freeing memory"
|
|
);
|
|
assert!(
|
|
metrics.recovery_time < Duration::from_secs(1),
|
|
"Recovery should be fast"
|
|
);
|
|
|
|
info!("Memory recovery time: {:?}", metrics.recovery_time);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_graceful_degradation_under_limits() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
// Test that system degrades gracefully, not catastrophically
|
|
let test = ThreadPoolExhaustionTest::new(1000, Duration::from_secs(10));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(metrics.graceful_degradation, "Should degrade gracefully");
|
|
assert!(metrics.system_stable, "System should not crash");
|
|
|
|
// Even under extreme load, some operations should succeed
|
|
let success_rate = metrics.success_rate();
|
|
assert!(
|
|
success_rate > 10.0,
|
|
"Should maintain >10% success rate: {:.2}%",
|
|
success_rate
|
|
);
|
|
|
|
info!("Success rate under load: {:.2}%", success_rate);
|
|
}
|
|
|
|
#[tokio::test]
|
|
#[ignore = "Very resource intensive - run manually"]
|
|
async fn test_multiple_resource_limits_simultaneously() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
info!("Running simultaneous resource limit test");
|
|
|
|
let duration = Duration::from_secs(30);
|
|
|
|
// Create test instances
|
|
let fd_test = FileDescriptorExhaustionTest::new(500, duration);
|
|
let thread_test = ThreadPoolExhaustionTest::new(500, duration);
|
|
let tcp_test = TcpConnectionLimitTest::new(250, duration, 19003);
|
|
let mem_test = MemoryAllocationLimitTest::new(200, 10, duration);
|
|
|
|
// Run all tests concurrently
|
|
let (fd_result, thread_result, tcp_result, mem_result) = tokio::join!(
|
|
fd_test.run(),
|
|
thread_test.run(),
|
|
tcp_test.run(),
|
|
mem_test.run(),
|
|
);
|
|
|
|
let fd_metrics = fd_result.expect("FD test failed");
|
|
let thread_metrics = thread_result.expect("Thread test failed");
|
|
let tcp_metrics = tcp_result.expect("TCP test failed");
|
|
let mem_metrics = mem_result.expect("Memory test failed");
|
|
|
|
// All tests should complete without crashes
|
|
assert!(fd_metrics.system_stable, "FD test should be stable");
|
|
assert!(thread_metrics.system_stable, "Thread test should be stable");
|
|
assert!(tcp_metrics.system_stable, "TCP test should be stable");
|
|
assert!(mem_metrics.system_stable, "Memory test should be stable");
|
|
|
|
info!("All resource limits tested simultaneously - system remained stable");
|
|
info!("Results:");
|
|
info!(" Files: {}", fd_metrics.successful_operations);
|
|
info!(" Threads: {}", thread_metrics.successful_operations);
|
|
info!(" TCP: {}", tcp_metrics.successful_operations);
|
|
info!(" Memory: {} MB", mem_metrics.peak_resource_usage);
|
|
}
|
|
|
|
#[tokio::test]
|
|
#[ignore = "Very resource intensive - run manually"]
|
|
async fn test_extreme_file_descriptor_pressure() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
// Try to open 10K files (should hit ulimit)
|
|
let test = FileDescriptorExhaustionTest::new(10_000, Duration::from_secs(60));
|
|
let metrics = test.run().await.expect("Test failed");
|
|
|
|
assert!(
|
|
metrics.graceful_degradation,
|
|
"Should hit file descriptor limit"
|
|
);
|
|
assert!(
|
|
metrics.system_stable,
|
|
"System should remain stable even at limit"
|
|
);
|
|
assert!(metrics.recovery_successful, "Should recover after limit");
|
|
|
|
info!(
|
|
"Extreme FD test: opened {} files",
|
|
metrics.successful_operations
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
#[ignore = "Very resource intensive - run manually"]
|
|
async fn test_sustained_resource_pressure() {
|
|
let _ = tracing_subscriber::fmt::try_init();
|
|
|
|
// Sustain high resource usage for extended period
|
|
let duration = Duration::from_secs(120);
|
|
|
|
let fd_test = FileDescriptorExhaustionTest::new(2000, duration);
|
|
let thread_test = ThreadPoolExhaustionTest::new(1000, duration);
|
|
|
|
let (fd_result, thread_result) = tokio::join!(fd_test.run(), thread_test.run(),);
|
|
|
|
let fd_metrics = fd_result.expect("FD test failed");
|
|
let thread_metrics = thread_result.expect("Thread test failed");
|
|
|
|
// After sustained pressure, system should still be stable
|
|
assert!(fd_metrics.system_stable, "System should remain stable");
|
|
assert!(thread_metrics.system_stable, "System should remain stable");
|
|
|
|
info!("Sustained pressure test complete - system stable");
|
|
}
|
|
}
|