fix(api): update system router prefix and test paths

- Add /system prefix to system router
- Update test paths to match new router prefix
- Remove duplicate root endpoint
- Update API documentation in README

Author: SQuent

README.md

@@ -1 +1,165 @@
-docker build -t stressed-pod .  && docker run -p 5000:5000 stressed-pod
+# Stressed Pod
+
+## Overview
+Stressed Pod is a tool designed to simulate various types of loads on a Kubernetes pod. It enables testing application resilience by generating CPU, memory loads, and producing custom logs. This tool is particularly useful for performance testing, resource sizing, and scaling policy validation.
+
+## Features
+- **CPU Load Management**: Precise control of CPU usage (0 to N cores)
+- **Memory Load Management**: Memory consumption simulation (in MB)
+- **Dynamic Load**: Progressive variation of CPU/memory load
+- **Log Generation**: Custom log production with different levels and formats
+- **Kubernetes Probes**: Control of readiness/liveness probes
+- **REST API**: HTTP interface for load control
+- **Environment Variable Configuration**: Flexible behavior parameterization
+
+## Installation
+
+### Requirements
+- Python 3.13+
+- Poetry for dependency management
+
+### Configuration
+
+#### Environment Variables
+```yaml
+# CPU Load
+ENABLE_DYNAMIC_CPU_LOAD: "false"
+INITIAL_CPU_LOAD: "0"
+FINAL_CPU_LOAD: "0.5"
+CPU_LOAD_DURATION: "60"
+STOP_CPU_LOAD_AT_END: "true"
+
+# Memory Load
+ENABLE_DYNAMIC_MEMORY_LOAD: "false"
+INITIAL_MEMORY_LOAD: "0"
+FINAL_MEMORY_LOAD: "256"
+MEMORY_LOAD_DURATION: "60"
+STOP_MEMORY_LOAD_AT_END: "true"
+
+# Log Configuration
+ENABLE_AUTOMATIC_LOGS: "false"
+LOG_MESSAGE: "Automatic log message"
+LOG_LEVEL: "info"
+LOG_SERVICE: "auto-logger"
+LOG_FORMAT: "json"
+LOG_INTERVAL: "5"
+LOG_DURATION: "60"
+
+# Initial Probe States
+READINESS_STATUS: "SUCCESS"
+LIVENESS_STATUS: "SUCCESS"
+
+# System Configuration
+ENABLE_AUTO_TERMINATION: "false"
+AUTO_TERMINATION_DELAY: "300"
+```
+
+## API Endpoints
+
+### Load Management
+- `GET /load`: Current load status
+- `POST /load/cpu/start`: Start CPU load
+- `POST /load/cpu/stop`: Stop CPU load
+- `POST /load/cpu/dynamic`: Configure dynamic CPU load
+- `POST /load/memory/start`: Start memory load
+- `POST /load/memory/stop`: Stop memory load
+- `POST /load/memory/dynamic`: Configure dynamic memory load
+
+### Log Management
+- `POST /log`: Create custom logs
+
+### Probe Management
+- `GET /probes`: Probe status
+- `GET /probes/readiness`: Readiness probe status
+- `GET /probes/liveness`: Liveness probe status
+- `POST /probes/status`: Modify probe status
+
+### System Management
+- `GET /system`: System information
+- `POST /system/terminate`: Schedule pod termination
+
+## Usage Examples
+
+### Dynamic CPU Load
+```bash
+curl -X POST http://localhost:8000/load/cpu/dynamic \
+  -H "Content-Type: application/json" \
+  -d '{
+    "start_value": 0.1,
+    "end_value": 0.8,
+    "duration": 60,
+    "stop_at_end": true
+  }'
+```
+
+### Memory Load
+```bash
+curl -X POST http://localhost:8000/load/memory/start \
+  -H "Content-Type: application/json" \
+  -d '{
+    "value": 256
+  }'
+```
+
+### Log Generation
+```bash
+curl -X POST http://localhost:8000/log \
+  -H "Content-Type: application/json" \
+  -d '{
+    "message": "Test message",
+    "level": "info",
+    "service": "test-service",
+    "format": "json",
+    "interval": 5,
+    "duration": 60
+  }'
+```
+
+### Probe Control
+```bash
+curl -X POST http://localhost:8000/probes/status \
+  -H "Content-Type: application/json" \
+  -d '{
+    "probe": "readiness",
+    "status": "error"
+  }'
+```
+
+## Development
+
+### Launch app
+docker-compose up --build
+
+### Running Tests
+```bash
+# Install dependencies
+poetry install
+
+# Run tests
+poetry run pytest
+
+# Run tests with coverage
+poetry run pytest --cov=app
+```
+
+### Code Quality
+The project uses:
+- pytest for testing
+- black for code formatting
+- flake8 for linting
+- mypy for type checking
+
+## Contributing
+Contributions are welcome! Please feel free to submit a Pull Request.
+
+## License
+This project is licensed under the MIT License.
+
+## Important Notes
+- The CPU load is distributed across all available cores
+- Memory load is specified in MB
+- Log formats supported: JSON and plaintext
+- Probe status changes are immediate
+- All durations are in seconds
+
+This tool is designed for testing purposes and should be used with caution in production environments.

app/main.py

@@ -2,8 +2,8 @@
 from .routers import load_router, probes_router, system_router, log_router
 
 app = FastAPI(
-    title="Load Testing API",
-    description="API for managing CPU and memory loads, and lifecycle probes",
+    title="Stressed API",
+    description="The API simulates controlled workloads and failures to stress-test a system, assessing its resilience, performance, and recovery capabilities.",
     version="1.0.0",
 )
 

app/managers/load_manager.py

@@ -2,6 +2,7 @@
 import signal
 import subprocess
 from threading import Timer
+import psutil
 
 
 class LoadManager:
@@ -28,7 +29,10 @@ def __init__(self):
         self.cpu_duration = int(os.getenv("CPU_LOAD_DURATION", 60))
         self.stop_cpu_at_end = os.getenv("STOP_CPU_LOAD_AT_END", "true") == "true"
 
-        # Call dynamic load functions during initialization if enabled
+        self.max_duration = 3600
+        self.system_memory = psutil.virtual_memory().total // (1024 * 1024)
+        self.system_cpus = os.cpu_count()
+
         if os.getenv("ENABLE_DYNAMIC_MEMORY_LOAD", "false") == "true":
             self.dynamic_memory_load(
                 self.memory_at_start,
@@ -62,10 +66,12 @@ def stop_cpu_load(self):
         self.cpu_timers.clear()
         self.cpu_requested = 0
 
-    def add_cpu_load(self, value: int):
-        """Add CPU load"""
-        if value < 0:
+    def add_cpu_load(self, value: float):
+        """Add CPU load with validation"""
+        if value <= 0:
             raise ValueError("CPU load must be greater than 0")
+        if value > self.system_cpus:
+            raise ValueError(f"CPU load cannot exceed system CPU count ({self.system_cpus})")
 
         if not os.path.exists(self.cpu_script_path):
             raise RuntimeError("CPU stress script is missing")
@@ -93,9 +99,11 @@ def stop_memory_load(self):
                 raise RuntimeError(f"Failed to terminate memory stress: {e}")
 
     def add_memory_load(self, value: int):
-        """Add memory load"""
-        if value < 0:
+        """Add memory load with validation"""
+        if value <= 0:
             raise ValueError("Memory load must be greater than 0")
+        if value > self.system_memory:
+            raise ValueError(f"Memory load cannot exceed system memory ({self.system_memory}MB)")
 
         if not os.path.exists(self.memory_script_path):
             raise RuntimeError("Memory stress script is missing")
@@ -111,9 +119,11 @@ def add_memory_load(self, value: int):
     def dynamic_memory_load(
         self, start_value: int, end_value: int, duration: int, stop_at_end: bool = False
     ):
-        """Dynamic memory load"""
-        if duration < 0:
-            raise ValueError("Duration must be greater than 0")
+        """Dynamic memory load with validation"""
+        if duration <= 0 or duration > self.max_duration:
+            raise ValueError(f"Duration must be between 1 and {self.max_duration} seconds")
+        if end_value < start_value:
+            raise ValueError("End value must be greater than start value")
 
         try:
             start_value = start_value
@@ -149,11 +159,13 @@ def apply_dynamic_memory_load(interval_num):
         apply_dynamic_memory_load(0)
 
     def dynamic_cpu_load(
-        self, start_value: int, end_value: int, duration: int, stop_at_end: bool = False
+        self, start_value: float, end_value: float, duration: int, stop_at_end: bool = False
     ):
-        """Dynamic CPU load"""
-        if duration < 0:
-            raise ValueError("Duration must be greater than 0")
+        """Dynamic CPU load with validation"""
+        if duration <= 0 or duration > self.max_duration:
+            raise ValueError(f"Duration must be between 1 and {self.max_duration} seconds")
+        if end_value < start_value:
+            raise ValueError("End value must be greater than start value")
 
         try:
             start_value = start_value
@@ -183,7 +195,6 @@ def apply_dynamic_cpu_load(interval_num):
                 timer.start()
 
             elif stop_at_end:
-                # Correctly schedule stop_memory_load to execute after the last interval
                 stop_timer = Timer(10, self.stop_cpu_load)
                 stop_timer.start()
 

app/managers/log_manager.py

@@ -1,13 +1,14 @@
 import logging
 import json
 from datetime import datetime, UTC
+import time
+import os
 from ..models.schemas import LogRequest, LogFormat
 import asyncio
-import time
 
 
 class JsonFormatter(logging.Formatter):
-    """Formatter personnalisé pour les logs JSON"""
+    """Custom formatter for JSON logs"""
 
     def format(self, record):
         return json.dumps(
@@ -21,7 +22,7 @@ def format(self, record):
 
 
 class PlainTextFormatter(logging.Formatter):
-    """Formatter personnalisé pour le texte plat"""
+    """Custom formatter for plain text logs"""
 
     def format(self, record):
         return f"{self.formatTime(record)} | {record.levelname} | {getattr(record, 'service', '-')} | {record.getMessage()}"
@@ -36,11 +37,14 @@ def __init__(self):
         self.json_formatter = JsonFormatter()
 
         self._setup_handler(LogFormat.JSON)
-
         self.current_format = LogFormat.JSON
 
+        # Start automatic logging if enabled
+        if os.getenv("ENABLE_AUTOMATIC_LOGS", "false").lower() == "true":
+            self._start_automatic_logs()
+
     def _setup_handler(self, format_type: LogFormat):
-        """Configure un nouveau handler avec le format spécifié"""
+        """Configure a new handler with the specified format"""
         for handler in self.logger.handlers[:]:
             self.logger.removeHandler(handler)
 
@@ -73,6 +77,29 @@ def _validate_interval_duration(self, interval: int | None, duration: int | None
         if interval and duration and interval > duration:
             raise ValueError("Interval cannot be greater than duration")
 
+    def _start_automatic_logs(self):
+        """Initialize automatic logging based on environment variables"""
+        log_data = LogRequest(
+            message=os.getenv("LOG_MESSAGE", "Automatic log message"),
+            level=os.getenv("LOG_LEVEL", "info").lower(),
+            service=os.getenv("LOG_SERVICE", "auto-logger"),
+            format=LogFormat(os.getenv("LOG_FORMAT", "json").lower()),
+            interval=int(os.getenv("LOG_INTERVAL", "5")),
+            duration=int(os.getenv("LOG_DURATION", "60"))
+        )
+        
+        # Create single log immediately
+        asyncio.create_task(self._create_single_log(
+            log_data, 
+            getattr(logging, log_data.level.upper())
+        ))
+
+        # Setup recurring logs if needed
+        if log_data.interval and log_data.duration:
+            asyncio.create_task(
+                self._create_recurring_logs(log_data, log_data.interval, log_data.duration)
+            )
+
     async def create_log(self, log_data: LogRequest) -> dict | str:
         """Create log"""
         level = getattr(logging, log_data.level.upper())
@@ -96,7 +123,7 @@ async def create_log(self, log_data: LogRequest) -> dict | str:
     async def _create_recurring_logs(
         self, log_data: LogRequest, interval: int, duration: int
     ):
-        """Crée des logs de manière récurrente"""
+        """Create logs at regular intervals"""
         end_time = time.time() + duration
         while time.time() < end_time:
             await self._create_single_log(
@@ -105,7 +132,7 @@ async def _create_recurring_logs(
             await asyncio.sleep(interval)
 
     async def _create_single_log(self, log_data: LogRequest, level: int) -> dict | str:
-        """Crée un seul log"""
+        """Create a single log"""
         try:
             timestamp = datetime.now(UTC).isoformat()
 
@@ -138,8 +165,8 @@ def _validate_log_level(self, level: str) -> bool:
         return level.lower() in valid_levels
 
     async def log_creator(self, log_data: LogRequest, interval: int, duration: int):
-        """Ffunction to create multiple log log"""
+        """Function to create multiple logs"""
         end_time = time.time() + duration
         while time.time() < end_time:
             await self.create_log(log_data)
-            await asyncio.sleep(interval)
+            await asyncio.sleep(interval)

app/managers/system_manager.py

@@ -1,8 +1,15 @@
 import psutil
 from datetime import datetime, timedelta
+import os
+import asyncio
 
 
 class SystemManager:
+    def __init__(self):
+        if os.getenv("ENABLE_AUTO_TERMINATION", "false").lower() == "true":
+            delay = int(os.getenv("AUTO_TERMINATION_DELAY", "300"))
+            self.terminate(delay)
+    
     def get_system_info(self):
         """Get system information"""
         return {