Merge pull request #19 from SQuent/fix-autoliniting--precommit

Fix autoliniting  precommit

Author: SQuent

.releaserc.yml

@@ -10,8 +10,18 @@ plugins:
     - replacements:
       - files:
         - "pyproject.toml"
-        from: "version = \".*\""
-        to: "version = \"${nextRelease.version}\""
+        from: "version = '.*'"
+        to: "version = '${nextRelease.version}'"
+        results:
+          - file: "pyproject.toml"
+            hasChanged: true
+            numMatches: 1
+            numReplacements: 1
+        countMatches: true
+      - files:
+        - "pyproject.toml"
+        from: "version_variable = \\[\\s*\"pyproject.toml:version\"\\s*\\]"
+        to: "version_variable = [\"pyproject.toml:version\"]"
         results:
           - file: "pyproject.toml"
             hasChanged: true

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="simulates controlled workloads and failures to stress-test a system",
     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,14 @@ 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 +101,13 @@ 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 +123,13 @@ 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 +165,19 @@ 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 +207,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()