diff --git a/Makefile b/Makefile index 3e83a3abe1..edb3b47850 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: all init format_backend format lint build run_backend dev help tests coverage clean_python_cache clean_npm_cache clean_frontend_build clean_all run_clic +.PHONY: all init format_backend format lint build run_backend dev help tests coverage clean_python_cache clean_npm_cache clean_frontend_build clean_all run_clic load_test_setup load_test_setup_basic load_test_list_flows load_test_run load_test_langflow_quick load_test_stress load_test_example load_test_clean load_test_remote_setup load_test_remote_run load_test_help # Configurations VERSION=$(shell grep "^version" pyproject.toml | sed 's/.*\"\(.*\)\"$$/\1/') @@ -11,6 +11,7 @@ PYTHON_REQUIRED=$(shell grep '^requires-python[[:space:]]*=' pyproject.toml | se RED=\033[0;31m NC=\033[0m # No Color GREEN=\033[0;32m +YELLOW=\033[1;33m log_level ?= debug host ?= 0.0.0.0 @@ -198,7 +199,7 @@ fix_codespell: ## run codespell to fix spelling errors format_backend: ## backend code formatters @uv run ruff check . --fix - @uv run ruff format . --config pyproject.toml + @uv run ruff format . format: format_backend format_frontend ## run code formatters @@ -602,6 +603,188 @@ locust: ## run locust load tests (options: locust_users=10 locust_spawn_rate=1 l -f $$(basename "$(locust_file)"); \ fi +# Enhanced load testing targets with improved error handling and shapes +load_test_host ?= http://127.0.0.1:8000 +load_test_flow_id ?= 5523731d-5ef3-56de-b4ef-59b0a224fdbc +load_test_api_key ?= test +html ?= false + +load_test_ramp100: ## Run 100-user ramp load test (3min, 0->100 users @ 5/s). Options: html=true, load_test_host, load_test_flow_id, load_test_api_key + @echo "$(YELLOW)Running 100-user ramp load test (3 minutes)$(NC)" + @export FLOW_ID=$(load_test_flow_id) && \ + export API_KEY=$(load_test_api_key) && \ + export REQUEST_TIMEOUT=10 && \ + cd src/backend/tests/locust && \ + if [ "$(html)" = "true" ]; then \ + echo "$(GREEN)Generating HTML report: ramp100_test.html$(NC)"; \ + uv run locust -f locustfile_complex_serve.py --host $(load_test_host) --headless --html ramp100_test.html; \ + else \ + uv run locust -f locustfile_complex_serve.py --host $(load_test_host) --headless; \ + fi + +load_test_cliff: ## Find performance cliff with step ramp (5->50 users, 30s steps). Options: html=true, load_test_host, load_test_flow_id, load_test_api_key + @echo "$(YELLOW)Running step ramp to find performance cliff$(NC)" + @export FLOW_ID=$(load_test_flow_id) && \ + export API_KEY=$(load_test_api_key) && \ + export REQUEST_TIMEOUT=10 && \ + cd src/backend/tests/locust && \ + if [ "$(html)" = "true" ]; then \ + echo "$(GREEN)Generating HTML report: cliff_test.html$(NC)"; \ + uv run locust -f lfx_step_ramp.py --host $(load_test_host) --headless --html cliff_test.html; \ + else \ + uv run locust -f lfx_step_ramp.py --host $(load_test_host) --headless; \ + fi + +load_test_lfx_quick: ## Quick LFX load test (30 users, 60s). Options: html=true, load_test_host, load_test_flow_id, load_test_api_key + @echo "$(YELLOW)Running quick 30-user load test (60 seconds)$(NC)" + @export FLOW_ID=$(load_test_flow_id) && \ + export API_KEY=$(load_test_api_key) && \ + export REQUEST_TIMEOUT=10 && \ + cd src/backend/tests/locust && \ + if [ "$(html)" = "true" ]; then \ + echo "$(GREEN)Generating HTML report: quick_test.html$(NC)"; \ + uv run locust -f lfx_serve_locustfile.py --host $(load_test_host) --headless -u 30 -r 5 -t 60s --html quick_test.html; \ + else \ + uv run locust -f lfx_serve_locustfile.py --host $(load_test_host) --headless -u 30 -r 5 -t 60s; \ + fi + +###################### +# ENHANCED LOAD TESTING +###################### + +# Enhanced load testing system with API-based flow loading +load_test_setup: ## Set up load test environment with starter project flows + @echo "$(YELLOW)Setting up Langflow load test environment$(NC)" + @cd src/backend/tests/locust && uv run python langflow_setup_test.py --interactive + +load_test_setup_basic: ## Set up load test environment with Basic Prompting flow + @echo "$(YELLOW)Setting up load test environment with Basic Prompting flow$(NC)" + @cd src/backend/tests/locust && uv run python langflow_setup_test.py --flow "Basic Prompting" --save-credentials load_test_creds.json + +load_test_list_flows: ## List available starter project flows + @echo "$(YELLOW)Listing available starter project flows$(NC)" + @cd src/backend/tests/locust && uv run python langflow_setup_test.py --list-flows + +load_test_run: ## Run load test (automatically sets up if needed). Use FLOW_NAME="Flow Name" to specify flow + @echo "$(YELLOW)Running load test with enhanced error logging$(NC)" + @if [ ! -f "src/backend/tests/locust/load_test_creds.json" ]; then \ + echo "$(BLUE)No credentials found. Running automatic setup...$(NC)"; \ + if [ -z "$(FLOW_NAME)" ]; then \ + echo "$(CYAN)Available flows:$(NC)"; \ + cd src/backend/tests/locust && uv run python langflow_setup_test.py --list-flows; \ + echo "$(RED)Please specify a flow: make load_test_run FLOW_NAME=\"Basic Prompting\"$(NC)"; \ + exit 1; \ + else \ + echo "$(BLUE)Setting up with flow: $(FLOW_NAME)$(NC)"; \ + cd src/backend/tests/locust && uv run python langflow_setup_test.py --flow "$(FLOW_NAME)" --save-credentials load_test_creds.json; \ + fi \ + fi + @cd src/backend/tests/locust && \ + export API_KEY=$$(python -c "import json; print(json.load(open('load_test_creds.json'))['api_key'])") && \ + export FLOW_ID=$$(python -c "import json; print(json.load(open('load_test_creds.json'))['flow_id'])") && \ + uv run python langflow_run_load_test.py --headless --users 20 --duration 120 --no-start-langflow --html load_test_report.html --csv load_test_results + +load_test_langflow_quick: ## Quick Langflow load test (10 users, 30s) with HTML report (automatically sets up if needed). Use FLOW_NAME="Flow Name" to specify flow + @echo "$(YELLOW)Running quick Langflow load test with HTML report$(NC)" + @if [ ! -f "src/backend/tests/locust/load_test_creds.json" ]; then \ + echo "$(BLUE)No credentials found. Running automatic setup...$(NC)"; \ + if [ -z "$(FLOW_NAME)" ]; then \ + echo "$(CYAN)Available flows:$(NC)"; \ + cd src/backend/tests/locust && uv run python langflow_setup_test.py --list-flows; \ + echo "$(RED)Please specify a flow: make load_test_langflow_quick FLOW_NAME=\"Basic Prompting\"$(NC)"; \ + exit 1; \ + else \ + echo "$(BLUE)Setting up with flow: $(FLOW_NAME)$(NC)"; \ + cd src/backend/tests/locust && uv run python langflow_setup_test.py --flow "$(FLOW_NAME)" --save-credentials load_test_creds.json; \ + fi \ + fi + @cd src/backend/tests/locust && \ + export API_KEY=$$(python -c "import json; print(json.load(open('load_test_creds.json'))['api_key'])") && \ + export FLOW_ID=$$(python -c "import json; print(json.load(open('load_test_creds.json'))['flow_id'])") && \ + uv run python langflow_run_load_test.py --headless --users 10 --duration 30 --no-start-langflow --html quick_test_report.html + +load_test_stress: ## Stress test (100 users, 5 minutes) with comprehensive reporting (automatically sets up if needed). Use FLOW_NAME="Flow Name" to specify flow + @echo "$(YELLOW)Running stress test with comprehensive reporting$(NC)" + @if [ ! -f "src/backend/tests/locust/load_test_creds.json" ]; then \ + echo "$(BLUE)No credentials found. Running automatic setup...$(NC)"; \ + if [ -z "$(FLOW_NAME)" ]; then \ + echo "$(CYAN)Available flows:$(NC)"; \ + cd src/backend/tests/locust && uv run python langflow_setup_test.py --list-flows; \ + echo "$(RED)Please specify a flow: make load_test_stress FLOW_NAME=\"Basic Prompting\"$(NC)"; \ + exit 1; \ + else \ + echo "$(BLUE)Setting up with flow: $(FLOW_NAME)$(NC)"; \ + cd src/backend/tests/locust && uv run python langflow_setup_test.py --flow "$(FLOW_NAME)" --save-credentials load_test_creds.json; \ + fi \ + fi + @cd src/backend/tests/locust && \ + export API_KEY=$$(python -c "import json; print(json.load(open('load_test_creds.json'))['api_key'])") && \ + export FLOW_ID=$$(python -c "import json; print(json.load(open('load_test_creds.json'))['flow_id'])") && \ + uv run python langflow_run_load_test.py --headless --users 100 --spawn-rate 5 --duration 300 --no-start-langflow --html stress_test_report.html --csv stress_test_results --shape ramp100 + +load_test_example: ## Run complete example workflow (setup + test + reports) + @echo "$(YELLOW)Running complete load test example workflow$(NC)" + @cd src/backend/tests/locust && uv run python langflow_example_workflow.py --auto + +load_test_clean: ## Clean up load test files and credentials + @echo "$(YELLOW)Cleaning up load test files$(NC)" + @cd src/backend/tests/locust && rm -f *.json *.html *.csv *.log + @echo "$(GREEN)Load test files cleaned$(NC)" + +load_test_remote_setup: ## Set up load test for remote instance (requires LANGFLOW_HOST) + @if [ -z "$(LANGFLOW_HOST)" ]; then \ + echo "$(RED)Error: LANGFLOW_HOST environment variable required$(NC)"; \ + echo "$(YELLOW)Example: export LANGFLOW_HOST=https://your-remote-instance.com$(NC)"; \ + exit 1; \ + fi + @echo "$(YELLOW)Setting up load test for remote instance: $(LANGFLOW_HOST)$(NC)" + @cd src/backend/tests/locust && uv run python langflow_setup_test.py --host $(LANGFLOW_HOST) --flow "Basic Prompting" --save-credentials remote_test_creds.json + +load_test_remote_run: ## Run load test against remote instance (requires prior setup) + @if [ -z "$(LANGFLOW_HOST)" ]; then \ + echo "$(RED)Error: LANGFLOW_HOST environment variable required$(NC)"; \ + exit 1; \ + fi + @if [ ! -f "src/backend/tests/locust/remote_test_creds.json" ]; then \ + echo "$(RED)Error: No remote credentials found. Run 'make load_test_remote_setup' first$(NC)"; \ + exit 1; \ + fi + @echo "$(YELLOW)Running load test against remote instance: $(LANGFLOW_HOST)$(NC)" + @cd src/backend/tests/locust && \ + export API_KEY=$$(python -c "import json; print(json.load(open('remote_test_creds.json'))['api_key'])") && \ + export FLOW_ID=$$(python -c "import json; print(json.load(open('remote_test_creds.json'))['flow_id'])") && \ + uv run python langflow_run_load_test.py --host $(LANGFLOW_HOST) --no-start-langflow --headless --users 10 --spawn-rate 1 --duration 120 --html remote_test_report.html + +load_test_help: ## Show detailed load testing help + @echo "$(GREEN)Langflow Enhanced Load Testing System$(NC)" + @echo "" + @echo "$(YELLOW)Quick Start (Local):$(NC)" + @echo " 1. make load_test_setup_basic # Set up with Basic Prompting flow" + @echo " 2. make load_test_langflow_quick # Run quick Langflow test" + @echo " 3. Open quick_test_report.html # View results" + @echo "" + @echo "$(YELLOW)Remote Testing:$(NC)" + @echo " 1. export LANGFLOW_HOST=https://your-instance.com" + @echo " 2. make load_test_remote_setup # Set up for remote testing" + @echo " 3. make load_test_remote_run # Run test against remote instance" + @echo "" + @echo "$(YELLOW)Available Commands:$(NC)" + @echo " load_test_setup - Interactive flow selection setup" + @echo " load_test_setup_basic - Quick setup with Basic Prompting" + @echo " load_test_list_flows - List available starter flows" + @echo " load_test_run - Standard load test (25 users, 2 min)" + @echo " load_test_langflow_quick - Quick Langflow test (10 users, 30s)" + @echo " load_test_quick - Quick complex serve test (30 users, 60s)" + @echo " load_test_stress - Stress test (100 users, 5 min)" + @echo " load_test_example - Complete example workflow" + @echo " load_test_clean - Clean up generated files" + @echo "" + @echo "$(YELLOW)Generated Reports:$(NC)" + @echo " - *.html files - Interactive HTML reports" + @echo " - *_results_*.csv - Raw performance data" + @echo " - *_detailed_errors_*.log - Comprehensive error logs" + @echo " - *_error_summary_*.json - Error analysis" + ###################### # INCLUDE FRONTEND MAKEFILE ###################### diff --git a/pyproject.toml b/pyproject.toml index 5ee57ff889..1a0010cfad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -175,7 +175,7 @@ dev = [ "types-aiofiles>=24.1.0.20240626", "codeflash>=0.8.4", "hypothesis>=6.123.17", - "locust>=2.32.9", + "locust~=2.40.5", "pytest-rerunfailures>=15.0", "scrapegraph-py>=1.10.2", "pydantic-ai>=0.0.19", @@ -346,6 +346,34 @@ external = ["RUF027"] "SLF001", "BLE001", # allow broad-exception catching in tests ] +"src/backend/tests/locust/*" = [ + "D1", # Missing docstrings (CLI tools don't need full docstrings) + "T201", # Print statements (needed for CLI output) + "S603", # Subprocess calls (needed for running commands) + "S607", # Starting process with partial executable path + "S104", # Binding to all interfaces (needed for Langflow server) + "S105", # Hardcoded passwords (test credentials) + "FBT001", # Boolean-typed positional arguments (common in CLI) + "FBT002", # Boolean default arguments (common in CLI tools) + "TRY002", # Custom exceptions (generic exceptions OK for CLI) + "TRY003", # Long exception messages (descriptive errors for users) + "TRY300", # Consider moving to else block (return patterns) + "EM101", # String literals in exceptions (user-facing messages) + "EM102", # F-string literals in exceptions (user-facing messages) + "EXE001", # Shebang without executable (may be run via python) + "D415", # First line punctuation (CLI docstrings) + "F401", # Unused imports (imports for availability checking) + "PTH123", # Use Path.open (open() is fine for simple cases) + "PTH107", # Use Path.unlink (os.remove is fine for simple cases) + "PTH207", # Use Path.glob (glob.glob is fine for simple cases) + "B007", # Unused loop variables (tuple unpacking) + "PLW0602", # Global variable usage (needed for locust state) + "PLW0603", # Global variable updates (needed for locust state) + "DTZ005", # Datetime without timezone (local time is fine for logs) + "G004", # Logging f-strings (detailed error logging) + "SIM102", # Nested if statements (readability over optimization) + "E501", # Line too long (some CLI commands are naturally long) +] [tool.ruff.lint.flake8-builtins] builtins-allowed-modules = [ "io", "logging", "socket"] diff --git a/src/backend/tests/locust/README.md b/src/backend/tests/locust/README.md new file mode 100644 index 0000000000..2a82e8b623 --- /dev/null +++ b/src/backend/tests/locust/README.md @@ -0,0 +1,332 @@ +# Langflow Load Testing + +This directory contains comprehensive load testing tools for both Langflow and LFX APIs using Locust. + +## ๐Ÿ”ง **Two Testing Systems** + +### **Langflow API Testing** (Enhanced System) + +- Files: `langflow_*.py` +- Tests the main Langflow application API +- Includes automatic setup, real starter projects, and comprehensive error logging + +### **LFX API Testing** (Complex Serve) + +- Files: `lfx_*.py` +- Tests the LFX complex serve API +- Uses traditional load testing patterns + +## Features + +- **Automatic Environment Setup**: Creates users, API keys, and test flows automatically +- **Pre-flight Testing**: Validates connectivity and flow execution before load testing +- **Multiple User Types**: Different user behaviors to simulate realistic load patterns +- **Load Test Shapes**: Predefined load patterns for different testing scenarios +- **Comprehensive Metrics**: Performance grading and detailed reporting +- **Enhanced Error Logging**: Detailed connection error analysis and Langflow log capture +- **Easy Setup**: One-command execution with automatic Langflow startup + +## Quick Start + +### Prerequisites + +```bash +# Using uv (recommended) +uv add locust httpx + +# Or using pip +pip install locust httpx +``` + +### Using Makefile Commands (Recommended) + +The easiest way to use the load testing system: + +```bash +# 1. Setup test environment (interactive flow selection) +make load_test_setup + +# 2. Run a quick test +make load_test_langflow_quick + +# 3. Run a full load test +make load_test_run + +# 4. Clean up when done +make load_test_clean + +# See all available commands +make load_test_help +``` + +## ๐ŸŒ **Remote Instance Testing** + +For testing against a remote Langflow instance: + +### Setup for Remote Testing + +```bash +# Using Makefile (recommended) +make load_test_remote_setup LANGFLOW_HOST="https://your-remote-instance.com" +make load_test_remote_run LANGFLOW_HOST="https://your-remote-instance.com" + +# Or using Python scripts directly +python langflow_setup_test.py --host https://your-remote-instance.com --interactive +python langflow_run_load_test.py --host https://your-remote-instance.com --no-start-langflow --headless --users 10 --duration 120 + +# Test remote instance before setup (optional) +python diagnose_remote.py --host https://your-remote-instance.com --load-test 5 +``` + +### Important Notes for Remote Testing + +- **Always use `--no-start-langflow`** when testing remote instances +- **Use HTTPS** for production remote instances +- **Consider network latency** in your performance expectations +- **Monitor both client and server resources** during testing +- **Use realistic user counts** based on your remote instance specs +- **Start with fewer users** (10 instead of 20) to avoid overwhelming remote instances +- **Use slower spawn rates** (1 user/sec) for more realistic load patterns +- **Status code 0 errors** indicate connection failures, usually from overloading the remote instance + +### Two-Step Process + +#### Step 1: Setup (Run Once) + +Choose and set up a real Langflow starter project for testing: + +```bash +# Interactive flow selection +python langflow_setup_test.py --interactive + +# Use specific flow +python langflow_setup_test.py --flow "Memory Chatbot" + +# List available flows +python langflow_setup_test.py --list-flows +``` + +This will: + +- Use default Langflow credentials (langflow/langflow) +- Generate API keys +- Upload a real starter project flow +- Provide credentials for load testing + +#### Step 2: Run Load Tests + +```bash +# Interactive mode with web UI +python langflow_run_load_test.py + +# Headless mode with 20 users for 2 minutes +python langflow_run_load_test.py --headless --users 20 --duration 120 + +# Use predefined load shape +python langflow_run_load_test.py --shape ramp100 --headless --users 100 --duration 180 +``` + +### Advanced Usage + +```bash +# Setup with custom host (e.g., remote instance) +python langflow_setup_test.py --host https://your-remote-instance.com --interactive + +# Save credentials to file +python langflow_setup_test.py --interactive --save-credentials my_test_creds.json + +# Test against existing remote Langflow instance +python langflow_run_load_test.py --host https://your-remote-instance.com --no-start-langflow + +# Save results to CSV and HTML +python langflow_run_load_test.py --headless --csv results --html report.html --users 50 --duration 300 + +# Direct Locust usage (after setup) +export API_KEY="your-api-key-from-setup" +export FLOW_ID="your-flow-id-from-setup" +locust -f langflow_locustfile.py --host http://localhost:7860 + +# Distributed testing (master) +locust -f langflow_locustfile.py --host http://localhost:7860 --master + +# Distributed testing (worker) +locust -f langflow_locustfile.py --host http://localhost:7860 --worker --master-host=localhost +``` + +## User Types + +The load test includes multiple user types that simulate different usage patterns: + +- **NormalUser** (default): Typical user behavior with realistic message distribution +- **AggressiveUser**: High-frequency requests with minimal wait times +- **SustainedLoadUser**: Constant 1 RPS per user for steady load testing +- **TailLatencyHunter**: Mixed workload to expose tail latency issues +- **ScalabilityTestUser**: Tests for scaling limits and performance cliffs +- **BurstUser**: Sends bursts of requests to test connection pooling + +## Load Test Shapes + +Predefined load patterns for different testing scenarios: + +- **RampToHundred**: 0 โ†’ 100 users over 20 seconds, hold for 3 minutes +- **StepRamp**: Step increases every 30 seconds to find performance cliffs + +Use with: `--shape ramp100` or `--shape stepramp` + +## Environment Variables + +- `LANGFLOW_HOST`: Base URL for Langflow server (default: http://localhost:7860) +- `SHAPE`: Load test shape (ramp100, stepramp) +- `REQUEST_TIMEOUT`: Request timeout in seconds (default: 30.0) + +## Architecture + +### Setup Process (`langflow_setup_test.py`) + +1. **Health Check**: Verify Langflow is running +2. **Flow Selection**: Choose from 40+ real starter project flows +3. **Authentication**: Login with default credentials (langflow/langflow) +4. **API Key Generation**: Create API key for load testing +5. **Flow Upload**: Upload the selected starter project flow +6. **Credential Export**: Provide environment variables for testing + +### Real Starter Project Flows + +Instead of simple test flows, the system uses real Langflow starter projects: + +- **Basic Prompting**: Simple LLM interaction +- **Memory Chatbot**: Conversational AI with memory +- **Document Q&A**: RAG-based document questioning +- **Research Agent**: Multi-step research workflows +- **Vector Store RAG**: Advanced retrieval-augmented generation +- **And 35+ more realistic flows** + +This provides much more realistic load testing scenarios that exercise: + +- Complex node graphs and dependencies +- Real LLM integrations +- Vector databases and embeddings +- Multi-step agent workflows +- Memory and state management + +## Performance Grading + +The load test provides automatic performance grading: + +- **Grade A**: Excellent performance, production ready +- **Grade B**: Good performance with minor issues +- **Grade C**: Acceptable but monitor closely +- **Grade D**: Performance issues detected +- **Grade F**: Significant problems, not production ready + +Grading is based on: + +- Failure rate (< 1% for A grade) +- 95th percentile response time (< 10s for good grades) +- Request throughput and consistency + +## Monitoring + +The test tracks: + +- Response times (p50, p95, p99) +- Request rates and throughput +- Failure rates and error types +- Slow requests (>10s, >20s) +- Connection and timeout issues + +## Troubleshooting + +### Common Issues + +1. **Setup Failed**: Ensure Langflow is accessible and not in read-only mode +2. **Authentication Errors**: Verify default credentials (langflow/langflow) are enabled +3. **Flow Creation Failed**: Verify the user has permission to create flows +4. **Connection Errors**: Check network connectivity and firewall settings +5. **Status Code 0 Errors**: Usually indicates connection overload - reduce user count or spawn rate + +### Debug Mode + +For debugging, you can: + +1. Run Langflow manually with `--log-level debug` +2. Check the Langflow logs for detailed error information +3. Use the web UI to verify the test flow was created correctly +4. Test API endpoints manually with curl or httpx +5. Use the diagnostic tool for remote instances: `python diagnose_remote.py --host --load-test 10` + +### Manual Setup + +If automatic setup fails, you can set up manually: + +1. Start Langflow: `python -m langflow run --auto-login` +2. Create a user account through the UI +3. Create an API key in the settings +4. Create a simple flow and note its ID +5. Set environment variables and run Locust directly + +```bash +export API_KEY="your-api-key" +export FLOW_ID="your-flow-id" +locust -f langflow_locustfile.py --host http://localhost:7860 +``` + +## Contributing + +When adding new user types or test scenarios: + +1. Inherit from `BaseLangflowUser` +2. Implement task methods with `@task` decorator +3. Use `self.make_request()` for consistent error handling +4. Add appropriate weight and wait_time settings +5. Document the user type's purpose and behavior + +## Examples + +### Basic Load Test + +```bash +python langflow_run_load_test.py --headless --users 10 --duration 60 +``` + +### Stress Test + +```bash +python langflow_run_load_test.py --shape ramp100 --headless --users 100 --duration 300 +``` + +### Performance Profiling + +```bash +python langflow_run_load_test.py --shape stepramp --headless --csv profile_results +``` + +### Production Readiness Test + +```bash +python langflow_run_load_test.py --users 50 --duration 600 --csv production_test --html production_report.html +``` + +## ๐Ÿ“Š HTML Reports + +The system generates beautiful HTML reports with: + +- **Interactive Charts**: Response time graphs, throughput charts +- **Detailed Statistics**: Request/response metrics, percentiles +- **Error Analysis**: Failure breakdown and error patterns +- **Performance Timeline**: Real-time performance during the test +- **User Behavior Analysis**: Breakdown by user type and task + +### HTML Report Examples + +```bash +# Generate comprehensive HTML report +python langflow_run_load_test.py --headless --users 25 --duration 120 --html detailed_report.html + +# Combined CSV + HTML reporting +python langflow_run_load_test.py --headless --users 100 --duration 300 --csv data --html analysis.html --shape ramp100 + +# Quick test with report +python langflow_setup_test.py --flow "Memory Chatbot" +python langflow_run_load_test.py --headless --users 10 --duration 60 --html quick_test.html +``` diff --git a/src/backend/tests/locust/diagnose_remote.py b/src/backend/tests/locust/diagnose_remote.py new file mode 100644 index 0000000000..1eabd0ed82 --- /dev/null +++ b/src/backend/tests/locust/diagnose_remote.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python3 +"""Diagnostic tool for remote Langflow instances. + +Helps debug connection issues and performance problems. +""" + +import argparse +import json +import sys +import time +from typing import Any + +import httpx + + +def test_connectivity(host: str) -> dict[str, Any]: + """Test basic connectivity to the host.""" + print(f"๐Ÿ”— Testing connectivity to {host}") + + results = { + "host": host, + "reachable": False, + "health_check": False, + "response_time_ms": None, + "error": None, + } + + try: + start_time = time.time() + with httpx.Client(timeout=30.0) as client: + response = client.get(f"{host}/health") + end_time = time.time() + + results["reachable"] = True + results["response_time_ms"] = round((end_time - start_time) * 1000, 2) + results["health_check"] = response.status_code == 200 + results["status_code"] = response.status_code + + if response.status_code == 200: + print(f" โœ… Health check passed ({results['response_time_ms']}ms)") + else: + print(f" โš ๏ธ Health check failed: {response.status_code}") + results["error"] = f"HTTP {response.status_code}" + + except Exception as e: + results["error"] = f"{type(e).__name__}: {e}" + print(f" โŒ Connection failed: {results['error']}") + + return results + + +def test_flow_endpoint(host: str, api_key: str, flow_id: str) -> dict[str, Any]: + """Test a flow execution request.""" + print("๐ŸŽฏ Testing flow execution") + + results = { + "success": False, + "response_time_ms": None, + "status_code": None, + "error": None, + "has_outputs": False, + } + + try: + url = f"{host}/api/v1/run/{flow_id}?stream=false" + payload = { + "input_value": "Hello, this is a diagnostic test", + "output_type": "chat", + "input_type": "chat", + "tweaks": {}, + } + headers = {"x-api-key": api_key, "Content-Type": "application/json"} + + start_time = time.time() + with httpx.Client(timeout=60.0) as client: + response = client.post(url, json=payload, headers=headers) + end_time = time.time() + + results["response_time_ms"] = round((end_time - start_time) * 1000, 2) + results["status_code"] = response.status_code + + if response.status_code == 200: + try: + data = response.json() + results["has_outputs"] = bool(data.get("outputs")) + results["success"] = results["has_outputs"] + + if results["success"]: + print(f" โœ… Flow execution successful ({results['response_time_ms']}ms)") + else: + print(f" โš ๏ธ Flow executed but no outputs ({results['response_time_ms']}ms)") + results["error"] = "No outputs in response" + + except Exception as e: + results["error"] = f"JSON decode error: {e}" + print(f" โŒ Invalid JSON response: {e}") + else: + results["error"] = f"HTTP {response.status_code}" + print(f" โŒ Flow execution failed: {response.status_code}") + print(f" Response: {response.text[:200]}...") + + except Exception as e: + results["error"] = f"{type(e).__name__}: {e}" + print(f" โŒ Request failed: {results['error']}") + + return results + + +def run_load_simulation(host: str, api_key: str, flow_id: str, num_requests: int = 10) -> dict[str, Any]: + """Run a small load simulation to test performance.""" + print(f"โšก Running mini load test ({num_requests} requests)") + + results = { + "total_requests": num_requests, + "successful_requests": 0, + "failed_requests": 0, + "connection_errors": 0, + "response_times": [], + "errors": [], + } + + url = f"{host}/api/v1/run/{flow_id}?stream=false" + payload = { + "input_value": "Load test message", + "output_type": "chat", + "input_type": "chat", + "tweaks": {}, + } + headers = {"x-api-key": api_key, "Content-Type": "application/json"} + + for i in range(num_requests): + try: + start_time = time.time() + with httpx.Client(timeout=30.0) as client: + response = client.post(url, json=payload, headers=headers) + end_time = time.time() + + response_time = round((end_time - start_time) * 1000, 2) + results["response_times"].append(response_time) + + if response.status_code == 200: + try: + data = response.json() + if data.get("outputs"): + results["successful_requests"] += 1 + print(f" โœ… Request {i + 1}: {response_time}ms") + else: + results["failed_requests"] += 1 + results["errors"].append(f"Request {i + 1}: No outputs") + print(f" โš ๏ธ Request {i + 1}: No outputs ({response_time}ms)") + except Exception as e: + results["failed_requests"] += 1 + results["errors"].append(f"Request {i + 1}: JSON error - {e}") + print(f" โŒ Request {i + 1}: JSON error ({response_time}ms)") + else: + results["failed_requests"] += 1 + results["errors"].append(f"Request {i + 1}: HTTP {response.status_code}") + print(f" โŒ Request {i + 1}: HTTP {response.status_code} ({response_time}ms)") + + except Exception as e: + results["connection_errors"] += 1 + results["errors"].append(f"Request {i + 1}: {type(e).__name__} - {e}") + print(f" ๐Ÿ’ฅ Request {i + 1}: Connection error - {e}") + + # Calculate statistics + if results["response_times"]: + results["avg_response_time"] = round(sum(results["response_times"]) / len(results["response_times"]), 2) + results["min_response_time"] = min(results["response_times"]) + results["max_response_time"] = max(results["response_times"]) + + return results + + +def main(): + parser = argparse.ArgumentParser(description="Diagnose remote Langflow instance") + parser.add_argument("--host", required=True, help="Langflow host URL") + parser.add_argument("--api-key", help="API key for flow execution") + parser.add_argument("--flow-id", help="Flow ID for testing") + parser.add_argument("--load-test", type=int, default=0, help="Number of requests for mini load test") + parser.add_argument("--output", help="Save results to JSON file") + + args = parser.parse_args() + + print(f"๐Ÿ” Diagnosing Langflow instance: {args.host}") + print("=" * 60) + + # Test basic connectivity + connectivity_results = test_connectivity(args.host) + + # Test flow execution if credentials provided + flow_results = None + if args.api_key and args.flow_id: + flow_results = test_flow_endpoint(args.host, args.api_key, args.flow_id) + else: + print("โš ๏ธ Skipping flow test (no API key or flow ID provided)") + + # Run mini load test if requested + load_results = None + if args.load_test > 0 and args.api_key and args.flow_id: + load_results = run_load_simulation(args.host, args.api_key, args.flow_id, args.load_test) + + # Summary + print("\n" + "=" * 60) + print("๐Ÿ“‹ DIAGNOSTIC SUMMARY") + print("=" * 60) + + print(f"Host: {args.host}") + print(f"Connectivity: {'โœ… OK' if connectivity_results['reachable'] else 'โŒ FAILED'}") + print(f"Health Check: {'โœ… OK' if connectivity_results['health_check'] else 'โŒ FAILED'}") + + if connectivity_results["response_time_ms"]: + print(f"Health Response Time: {connectivity_results['response_time_ms']}ms") + + if flow_results: + print(f"Flow Execution: {'โœ… OK' if flow_results['success'] else 'โŒ FAILED'}") + if flow_results["response_time_ms"]: + print(f"Flow Response Time: {flow_results['response_time_ms']}ms") + + if load_results: + success_rate = (load_results["successful_requests"] / load_results["total_requests"]) * 100 + print( + f"Mini Load Test: {load_results['successful_requests']}/{load_results['total_requests']} ({success_rate:.1f}% success)" + ) + if load_results.get("avg_response_time"): + print(f"Average Response Time: {load_results['avg_response_time']}ms") + + # Recommendations + print("\n๐Ÿ”ง RECOMMENDATIONS:") + if not connectivity_results["reachable"]: + print("โŒ Cannot reach the host - check URL and network connectivity") + elif not connectivity_results["health_check"]: + print("โŒ Health check failed - Langflow may not be running properly") + elif flow_results and not flow_results["success"]: + print("โŒ Flow execution failed - check API key, flow ID, and flow configuration") + elif load_results and load_results["connection_errors"] > 0: + print("โš ๏ธ Connection errors detected - instance may be overloaded or unstable") + elif load_results and load_results.get("avg_response_time", 0) > 10000: + print("โš ๏ธ Slow response times - consider reducing load or optimizing flow") + else: + print("โœ… Instance appears healthy for load testing") + + # Save results if requested + if args.output: + results = { + "timestamp": time.time(), + "host": args.host, + "connectivity": connectivity_results, + "flow_execution": flow_results, + "load_simulation": load_results, + } + + with open(args.output, "w") as f: + json.dump(results, f, indent=2) + print(f"\n๐Ÿ’พ Results saved to: {args.output}") + + +if __name__ == "__main__": + main() diff --git a/src/backend/tests/locust/langflow_example_workflow.py b/src/backend/tests/locust/langflow_example_workflow.py new file mode 100644 index 0000000000..dd04fe03a5 --- /dev/null +++ b/src/backend/tests/locust/langflow_example_workflow.py @@ -0,0 +1,284 @@ +#!/usr/bin/env python3 +"""Example Langflow Load Testing Workflow + +This script demonstrates the complete workflow for setting up and running +Langflow load tests with real starter project flows. + +Usage: + python example_workflow.py +""" + +import os +import subprocess +import sys +from pathlib import Path + + +def run_command(cmd, description="", check=True): + """Run a command with nice output.""" + print(f"\n{'=' * 60}") + print(f"RUNNING: {description}") + print(f"{'=' * 60}") + print(f"Command: {' '.join(cmd) if isinstance(cmd, list) else cmd}") + print() + + try: + result = subprocess.run(cmd, check=check) + return result.returncode == 0 + except subprocess.CalledProcessError as e: + print(f"โŒ Command failed with exit code {e.returncode}") + return False + except KeyboardInterrupt: + print("\nโš ๏ธ Interrupted by user") + return False + + +def run_command_with_env(cmd, description="", env=None, check=True): + """Run a command with custom environment variables.""" + print(f"\n{'=' * 60}") + print(f"RUNNING: {description}") + print(f"{'=' * 60}") + print(f"Command: {' '.join(cmd) if isinstance(cmd, list) else cmd}") + print() + + try: + result = subprocess.run(cmd, env=env, check=check) + return result.returncode == 0 + except subprocess.CalledProcessError as e: + print(f"โŒ Command failed with exit code {e.returncode}") + return False + except KeyboardInterrupt: + print("\nโš ๏ธ Interrupted by user") + return False + + +def check_dependencies(): + """Check if required dependencies are installed.""" + print("Checking dependencies...") + + try: + import httpx + + print("โœ… httpx is available") + except ImportError: + print("โŒ httpx not found. Install with: pip install httpx") + return False + + try: + result = subprocess.run(["locust", "--version"], check=False, capture_output=True, text=True) + if result.returncode == 0: + print(f"โœ… locust is available: {result.stdout.strip()}") + else: + print("โŒ locust not found. Install with: pip install locust") + return False + except FileNotFoundError: + print("โŒ locust not found. Install with: pip install locust") + return False + + return True + + +def main(): + import argparse + + parser = argparse.ArgumentParser(description="Example Langflow Load Testing Workflow") + parser.add_argument("--auto", action="store_true", help="Run automatically without user input prompts") + args = parser.parse_args() + + print("๐Ÿš€ Langflow Load Testing Example Workflow") + print("This example will demonstrate the complete load testing setup and execution.") + + # Check dependencies + if not check_dependencies(): + print("\nโŒ Missing dependencies. Please install them and try again.") + sys.exit(1) + + script_dir = Path(__file__).parent + setup_script = script_dir / "langflow_setup_test.py" + runner_script = script_dir / "langflow_run_load_test.py" + + # Check if scripts exist + if not setup_script.exists(): + print(f"โŒ Setup script not found: {setup_script}") + sys.exit(1) + + if not runner_script.exists(): + print(f"โŒ Runner script not found: {runner_script}") + sys.exit(1) + + print("\n" + "=" * 80) + print("EXAMPLE WORKFLOW STEPS") + print("=" * 80) + print("1. List available starter project flows") + print("2. Set up test environment with a selected flow") + print("3. Run a quick load test") + print("4. Show results and cleanup options") + print("=" * 80) + + def wait_for_user(message): + if args.auto: + print(f"\n{message} (auto mode - continuing...)") + import time + + time.sleep(1) + else: + input(f"\n{message}") + + try: + # Step 1: List available flows + wait_for_user("Press Enter to list available starter project flows...") + if not run_command([sys.executable, str(setup_script), "--list-flows"], "List available starter project flows"): + return + + # Step 2: Setup with Basic Prompting (good for examples) + wait_for_user("Press Enter to set up test environment with 'Basic Prompting' flow...") + if not run_command( + [ + sys.executable, + str(setup_script), + "--flow", + "Basic Prompting", + "--save-credentials", + "example_test_creds.json", + ], + "Set up test environment", + ): + return + + # Step 3: Run a quick load test + wait_for_user("Press Enter to run a quick load test (10 users, 30 seconds)...") + + # Load credentials from the saved file and set environment variables + try: + import json + + with open("example_test_creds.json") as f: + creds = json.load(f) + + # Set environment variables for the load test + env = os.environ.copy() + env["LANGFLOW_HOST"] = creds["host"] + env["API_KEY"] = creds["api_key"] + env["FLOW_ID"] = creds["flow_id"] + + print(" ๐Ÿ”ง Setting environment variables:") + print(f" LANGFLOW_HOST={creds['host']}") + print(f" API_KEY={creds['api_key'][:20]}...") + print(f" FLOW_ID={creds['flow_id']}") + + except Exception as e: + print(f" โš ๏ธ Could not load credentials: {e}") + env = os.environ.copy() + + # Run the load test with proper environment + success = run_command_with_env( + [ + sys.executable, + str(runner_script), + "--headless", + "--users", + "100", + "--spawn-rate", + "2", + "--duration", + "30", + "--no-start-langflow", + "--html", + "langflow_load_test_report.html", + "--csv", + "langflow_load_test_results", + ], + "Run quick load test with HTML report generation", + env=env, + ) + + if not success: + print("โš ๏ธ Load test may have failed, but that's okay for this example") + + # Step 4: Show what's possible + print(f"\n{'=' * 80}") + print("EXAMPLE COMPLETE - WHAT'S NEXT?") + print(f"{'=' * 80}") + print("The example workflow is complete! Here's what you can do next:") + print() + print("๐Ÿ”ง Try different flows:") + print(" python setup_langflow_test.py --interactive") + print() + print("๐Ÿ“Š Run more comprehensive tests:") + print(" python run_load_test.py --shape ramp100 --headless --users 100 --duration 180") + print() + print("๐ŸŒ Use the web UI for interactive testing:") + print(" python run_load_test.py") + print() + print("๐Ÿ’พ Your test credentials are saved in: example_test_creds.json") + print() + print("๐Ÿ“Š Generated Reports:") + print(" - langflow_load_test_report.html (detailed HTML report)") + print(" - langflow_load_test_results_*.csv (CSV data files)") + print(" - langflow_load_test_detailed_errors_*.log (detailed error logs)") + print(" - langflow_load_test_error_summary_*.json (error analysis)") + print(" - langflow_server_logs_during_test_*.log (Langflow server logs)") + print() + print("๐Ÿงน Clean up:") + print(" - Remove test flows from Langflow UI") + print(" - Delete example_test_creds.json") + print(" - Delete generated report files") + print(" - Reset environment variables") + print(f"{'=' * 80}") + + # Cleanup option + if args.auto: + print("\nAuto mode - skipping cleanup so you can view the generated reports!") + print("๐Ÿ“ Files preserved:") + print(" - example_test_creds.json") + print(" - langflow_load_test_report.html") + print(" - langflow_load_test_results_*.csv") + else: + cleanup_response = input("\nWould you like to clean up the example files? (y/N): ").strip().lower() + + if cleanup_response == "y": + files_to_clean = [ + "example_test_creds.json", + "langflow_load_test_report.html", + "langflow_load_test_results_failures.csv", + "langflow_load_test_results_stats.csv", + "langflow_load_test_results_stats_history.csv", + "langflow_load_test_results_exceptions.csv", + ] + + # Also clean up error logs (they have timestamps, so use glob pattern) + import glob + + error_files = glob.glob("langflow_load_test_detailed_errors_*.log") + error_files.extend(glob.glob("langflow_load_test_error_summary_*.json")) + error_files.extend(glob.glob("langflow_server_logs_during_test_*.log")) + files_to_clean.extend(error_files) + + for file_path in files_to_clean: + try: + os.remove(file_path) + print(f"โœ… Cleaned up {file_path}") + except FileNotFoundError: + pass # File doesn't exist, that's fine + except Exception as e: + print(f"โš ๏ธ Could not clean up {file_path}: {e}") + + print("\n๐ŸŽ‰ Example workflow completed successfully!") + print("You're now ready to use Langflow load testing for your own projects.") + print() + print("๐Ÿ“Š View your load test results:") + print(" โ€ข Open langflow_load_test_report.html in your browser for detailed analysis") + print(" โ€ข Check langflow_load_test_results_*.csv for raw data") + print(" โ€ข Review langflow_load_test_detailed_errors_*.log for comprehensive error details") + print(" โ€ข Analyze langflow_load_test_error_summary_*.json for error patterns") + print(" โ€ข Examine langflow_server_logs_during_test_*.log for server-side issues") + + except KeyboardInterrupt: + print("\n\nโš ๏ธ Example workflow interrupted by user") + except Exception as e: + print(f"\nโŒ Example workflow failed: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/src/backend/tests/locust/langflow_locustfile.py b/src/backend/tests/locust/langflow_locustfile.py new file mode 100644 index 0000000000..e066ed1c46 --- /dev/null +++ b/src/backend/tests/locust/langflow_locustfile.py @@ -0,0 +1,624 @@ +"""Langflow API Locust Load Testing File. + +Comprehensive load testing for Langflow API with multiple user behaviors and performance analysis. +Based on production-ready patterns with proper error handling, metrics tracking, and reporting. + +Usage: + # Run with web UI (recommended) + locust -f locustfile.py --host http://localhost:7860 + + # Run headless with built-in shape + locust -f locustfile.py --host http://localhost:7860 --headless --shape RampToHundred + + # Run distributed (master) + locust -f locustfile.py --host http://localhost:7860 --master + + # Run distributed (worker) + locust -f locustfile.py --host http://localhost:7860 --worker --master-host=localhost + +Environment Variables: + - LANGFLOW_HOST: Base URL for the Langflow server (default: http://localhost:7860) + - FLOW_ID: Flow ID to test (required) + - API_KEY: API key for authentication (required) + - MIN_WAIT: Minimum wait time between requests in ms (default: 2000) + - MAX_WAIT: Maximum wait time between requests in ms (default: 5000) + - REQUEST_TIMEOUT: Request timeout in seconds (default: 30.0) + - SHAPE: Load test shape to use (default: none, options: ramp100, stepramp) +""" + +import inspect +import json +import logging +import os +import random +import time +import traceback +from datetime import datetime +from pathlib import Path + +import gevent +from locust import FastHttpUser, LoadTestShape, between, constant, constant_pacing, events, task + +# Test messages with realistic distribution +TEST_MESSAGES = { + "minimal": "Hi", + "simple": "Can you help me?", + "medium": "I need help understanding how machine learning works in this context.", + "complex": "Please analyze this data: " + "x" * 500 + " and provide detailed insights.", + "large": "Here's a complex scenario: " + "data " * 1000, + "realistic": "Hey, Could you check https://docs.langflow.org for me? Later, could you calculate 1390 / 192 ?", +} + +# Weighted message distribution for realistic load +MESSAGE_WEIGHTS = [("simple", 40), ("realistic", 25), ("medium", 20), ("minimal", 10), ("complex", 4), ("large", 1)] + +# Enhanced error logging setup +ERROR_LOG_FILE = None +LANGFLOW_LOG_FILE = None +DETAILED_ERRORS = [] + + +def setup_error_logging(): + """Set up detailed error logging for the load test.""" + global ERROR_LOG_FILE, LANGFLOW_LOG_FILE + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + + # Create detailed error log + ERROR_LOG_FILE = f"langflow_load_test_detailed_errors_{timestamp}.log" + + # Set up error logger + error_logger = logging.getLogger("langflow_load_test_errors") + error_logger.setLevel(logging.DEBUG) + + # Create file handler + error_handler = logging.FileHandler(ERROR_LOG_FILE) + error_handler.setLevel(logging.DEBUG) + + # Create formatter + formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") + error_handler.setFormatter(formatter) + + error_logger.addHandler(error_handler) + + # Try to capture Langflow logs + langflow_log_paths = ["langflow.log", "logs/langflow.log", "../../../langflow.log", "../../../../langflow.log"] + + for log_path in langflow_log_paths: + if Path(log_path).exists(): + LANGFLOW_LOG_FILE = log_path + break + + print("๐Ÿ“ Error logging setup:") + print(f" โ€ข Detailed errors: {ERROR_LOG_FILE}") + if LANGFLOW_LOG_FILE: + print(f" โ€ข Langflow logs: {LANGFLOW_LOG_FILE}") + else: + print(" โ€ข Langflow logs: Not found (will monitor common locations)") + + +def log_detailed_error( + user_class, method, url, status_code, response_text, exception=None, request_data=None, traceback=None +): + """Log detailed error information.""" + global DETAILED_ERRORS + + error_logger = logging.getLogger("langflow_load_test_errors") + + error_info = { + "timestamp": datetime.now().isoformat(), + "user_class": user_class, + "method": method, + "url": url, + "status_code": status_code, + "response_text": response_text[:1000] if response_text else None, # Limit response size + "request_data": request_data, + "exception": str(exception) if exception else None, + "traceback": traceback if traceback else None, + } + + DETAILED_ERRORS.append(error_info) + + # Log to file + error_logger.error(f""" +=== LOAD TEST ERROR === +User Class: {user_class} +Method: {method} +URL: {url} +Status Code: {status_code} +Request Data: {json.dumps(request_data, indent=2) if request_data else "None"} +Response Text: {response_text[:500] if response_text else "None"}... +Exception: {exception} +Traceback: {traceback.format_exc() if exception else "None"} +======================== +""") + + +def save_error_summary(): + """Save a summary of all errors encountered during the test.""" + if not DETAILED_ERRORS: + return + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + summary_file = f"langflow_load_test_error_summary_{timestamp}.json" + + # Group errors by type + error_summary = {} + for error in DETAILED_ERRORS: + key = f"{error['status_code']}_{error['user_class']}" + if key not in error_summary: + error_summary[key] = { + "count": 0, + "examples": [], + "status_code": error["status_code"], + "user_class": error["user_class"], + } + + error_summary[key]["count"] += 1 + if len(error_summary[key]["examples"]) < 3: # Keep up to 3 examples + error_summary[key]["examples"].append(error) + + # Save summary + with open(summary_file, "w") as f: + json.dump( + { + "test_timestamp": timestamp, + "total_errors": len(DETAILED_ERRORS), + "error_types": len(error_summary), + "error_breakdown": error_summary, + }, + f, + indent=2, + ) + + print(f"๐Ÿ“Š Error summary saved: {summary_file}") + + +def capture_langflow_logs(): + """Capture recent Langflow logs if available.""" + if not LANGFLOW_LOG_FILE or not Path(LANGFLOW_LOG_FILE).exists(): + return None + + try: + # Read last 1000 lines of Langflow log + with open(LANGFLOW_LOG_FILE) as f: + lines = f.readlines() + recent_lines = lines[-1000:] if len(lines) > 1000 else lines + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + captured_log_file = f"langflow_server_logs_during_test_{timestamp}.log" + + with open(captured_log_file, "w") as f: + f.write("# Langflow server logs captured during load test\n") + f.write(f"# Original log file: {LANGFLOW_LOG_FILE}\n") + f.write(f"# Capture time: {datetime.now().isoformat()}\n") + f.write(f"# Lines captured: {len(recent_lines)}\n\n") + f.writelines(recent_lines) + + print(f"๐Ÿ“‹ Langflow logs captured: {captured_log_file}") + return captured_log_file + + except Exception as e: + print(f"โš ๏ธ Could not capture Langflow logs: {e}") + return None + + +# Load test shapes +class RampToHundred(LoadTestShape): + """0 -> 100 users at 5 users/sec (20s ramp), then hold until 180s total. + + Matches production testing patterns: 3 minutes, ramping to 100 users. + """ + + spawn_rate = 5 + target_users = 100 + total_duration = 180 # seconds + + def tick(self): + run_time = self.get_run_time() + if run_time >= self.total_duration: + return None + users = min(int(run_time * self.spawn_rate), self.target_users) + return users, self.spawn_rate + + +class StepRamp(LoadTestShape): + """Step ramp for finding performance cliffs. + + Steps every 30 seconds: 5 -> 10 -> 15 -> 20 -> 25 -> 30 -> 35 users. + Each step holds for exactly 30 seconds to measure steady-state performance. + """ + + def tick(self): + run_time = self.get_run_time() + + # Define the step progression with 30-second intervals + steps = [ + (30, 5), # 0-30s: 5 users + (60, 10), # 30-60s: 10 users + (90, 15), # 60-90s: 15 users + (120, 20), # 90-120s: 20 users + (150, 25), # 120-150s: 25 users + (180, 30), # 150-180s: 30 users + (210, 35), # 180-210s: 35 users + (240, 40), # 210-240s: 40 users + (270, 45), # 240-270s: 45 users + (300, 50), # 270-300s: 50 users + ] + + # Find current step + for time_limit, user_count in steps: + if run_time < time_limit: + return user_count, 10 # Fast spawn rate for quick transitions + + return None # End test after 300 seconds + + +# Environment-scoped metrics tracking +_env_bags = {} + + +@events.test_start.add_listener +def on_test_start(environment, **_kwargs): + """Initialize per-environment metrics tracking.""" + # Set up enhanced error logging + setup_error_logging() + + _env_bags[environment] = { + "slow_10s": 0, + "slow_20s": 0, + } + + +@events.request.add_listener +def on_request(request_type, name, response_time, response_length, exception, context, **kwargs): # noqa: ARG001 + """Track slow requests using Locust's built-in timing.""" + # response_time is in milliseconds from Locust + bag = _env_bags.get(context.get("environment") if context else None) + if bag is None: + # fallback: try the single environment we likely have + if len(_env_bags) == 1: + bag = next(iter(_env_bags.values())) + else: + return + + if exception is None: # Only count successful requests for timing + if response_time > 10_000: # 10 seconds in ms + bag["slow_10s"] += 1 + if response_time > 20_000: # 20 seconds in ms + bag["slow_20s"] += 1 + + +@events.test_stop.add_listener +def on_test_stop(environment, **_kwargs): + """Print comprehensive test summary with performance grading.""" + stats = environment.stats.total + if stats.num_requests == 0: + return + + # Get percentiles and basic stats + p50 = stats.get_response_time_percentile(0.50) or 0 + p95 = stats.get_response_time_percentile(0.95) or 0 + p99 = stats.get_response_time_percentile(0.99) or 0 + fail_ratio = stats.fail_ratio + current_rps = getattr(stats, "current_rps", 0.0) + + # Get slow request counts + bag = _env_bags.get(environment, {"slow_10s": 0, "slow_20s": 0}) + + # Performance grading based on production criteria + grade = "A" + issues = [] + + if fail_ratio > 0.01: + grade = "B" + issues.append(f"fail {fail_ratio:.1%}") + if fail_ratio > 0.05: + grade = "C" + if p95 > 10_000: + grade = max(grade, "D") + issues.append(f"p95 {p95 / 1000:.1f}s") + if p95 > 20_000: + grade = "F" + issues.append(f"p95 {p95 / 1000:.1f}s") + + print(f"\n{'=' * 60}") + print(f"LANGFLOW API LOAD TEST RESULTS - GRADE: {grade}") + print(f"{'=' * 60}") + print(f"Requests: {stats.num_requests:,} | Failures: {stats.num_failures:,} ({fail_ratio:.1%})") + print(f"Response Times: p50={p50 / 1000:.2f}s p95={p95 / 1000:.2f}s p99={p99 / 1000:.2f}s") + print(f"RPS: {current_rps:.1f} | Slow requests: >10s={bag['slow_10s']} >20s={bag['slow_20s']}") + + if issues: + print(f"Issues: {', '.join(issues)}") + + # Production readiness assessment + if grade in ["A", "B"]: + print("โœ… PRODUCTION READY - Performance meets production standards") + elif grade == "C": + print("โš ๏ธ CAUTION - Acceptable but monitor closely in production") + else: + print("โŒ NOT PRODUCTION READY - Significant performance issues detected") + + print(f"{'=' * 60}\n") + + # Save detailed error information + save_error_summary() + + # Capture Langflow logs + capture_langflow_logs() + + # Set exit code for CI/CD + if fail_ratio > 0.01: + environment.process_exit_code = 1 + + # Cleanup + _env_bags.pop(environment, None) + + +class BaseLangflowUser(FastHttpUser): + """Base class for all Langflow API load testing user types.""" + + abstract = True + REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "30.0")) + + # Use the host provided by environment variable or default + host = os.getenv("LANGFLOW_HOST", "http://localhost:7860") + + def on_start(self): + """Called when a user starts before any task is scheduled.""" + # Get credentials from environment variables + self.api_key = os.getenv("API_KEY") + self.flow_id = os.getenv("FLOW_ID") + + if not self.api_key: + raise ValueError("API_KEY environment variable is required. Run setup_langflow_test.py first.") + + if not self.flow_id: + raise ValueError("FLOW_ID environment variable is required. Run setup_langflow_test.py first.") + + self.session_id = f"locust_{self.__class__.__name__}_{id(self)}_{int(time.time())}" + self.request_count = 0 + + # Test connection and auth + with self.client.get("/health", catch_response=True) as response: + if response.status_code != 200: + raise ConnectionError(f"Health check failed: {response.status_code}") + + def make_request(self, message_type="simple", tag_suffix=""): + """Make a request with proper error handling and timing.""" + message = TEST_MESSAGES.get(message_type, TEST_MESSAGES["simple"]) + + # Langflow API payload structure + payload = { + "input_value": message, + "output_type": "chat", + "input_type": "chat", + "tweaks": {}, + } + + headers = { + "x-api-key": self.api_key, + "Content-Type": "application/json", + "Accept": "application/json", + } + + self.request_count += 1 + endpoint = f"/api/v1/run/{self.flow_id}?stream=false" + name = f"{endpoint} [{message_type}{tag_suffix}]" + + try: + with self.client.post( + endpoint, + json=payload, + headers=headers, + name=name, + timeout=self.REQUEST_TIMEOUT, + catch_response=True, + ) as response: + # Get response text for error logging + try: + response_text = response.text + except Exception: + response_text = "Could not read response text" + + # Handle successful responses + if response.status_code == 200: + try: + data = response.json() + # Langflow API success check - look for outputs + if data.get("outputs"): + return response.success() + # Check for error messages in the response + error_msg = data.get("detail", "Unknown error") + + # Log detailed error for successful HTTP but failed flow execution + log_detailed_error( + user_class=self.__class__.__name__, + method="POST", + url=f"{self.host}{endpoint}", + status_code=response.status_code, + response_text=response_text, + request_data=payload, + exception=None, + ) + + return response.failure(f"Flow execution failed: {error_msg}") + except json.JSONDecodeError as e: + log_detailed_error( + user_class=self.__class__.__name__, + method="POST", + url=f"{self.host}{endpoint}", + status_code=response.status_code, + response_text=response_text, + request_data=payload, + exception=e, + ) + return response.failure("Invalid JSON response") + + # Log all error responses with detailed information + log_detailed_error( + user_class=self.__class__.__name__, + method="POST", + url=f"{self.host}{endpoint}", + status_code=response.status_code, + response_text=response_text, + request_data=payload, + exception=None, + ) + + # Handle specific error cases + if response.status_code in (429, 503): + return response.failure(f"Backpressure/capacity: {response.status_code}") + if response.status_code == 401: + return response.failure("Unauthorized - API key issue") + if response.status_code == 404: + return response.failure("Flow not found - check FLOW_ID") + if response.status_code >= 500: + return response.failure(f"Server error {response.status_code}") + + return response.failure(f"HTTP {response.status_code}") + + except Exception as e: + # Get more detailed error information + error_details = { + "error_type": type(e).__name__, + "error_message": str(e), + "is_timeout": "timeout" in str(e).lower(), + "is_connection_error": "connection" in str(e).lower(), + "is_dns_error": "name resolution" in str(e).lower() or "dns" in str(e).lower(), + } + + # Log any exceptions that occur during the request + log_detailed_error( + user_class=self.__class__.__name__, + method="POST", + url=f"{self.host}{endpoint}", + status_code=0, # Connection error + response_text=f"Connection Error: {error_details}", + request_data=payload, + exception=str(e), + traceback=traceback.format_exc(), + ) + # Re-raise the exception so Locust can handle it properly + raise + + +class NormalUser(BaseLangflowUser): + """Normal user simulating typical API interactions.""" + + weight = 3 + wait_time = between(0.5, 2) # Typical user think time + + @task(80) + def send_message(self): + """Main task: Send a message with weighted distribution.""" + message_type = random.choices([w[0] for w in MESSAGE_WEIGHTS], weights=[w[1] for w in MESSAGE_WEIGHTS], k=1)[0] # noqa: S311 + self.make_request(message_type=message_type) + + @task(15) + def send_burst(self): + """Send a burst of 3 small messages quickly.""" + for i in range(3): + self.make_request(message_type="minimal", tag_suffix=f"-burst{i}") + gevent.sleep(0.1) # Small delay between burst requests + + @task(5) + def send_complex(self): + """Occasionally send complex requests that stress the system.""" + self.make_request(message_type="complex") + + +class AggressiveUser(BaseLangflowUser): + """Aggressive user with minimal wait times.""" + + weight = 3 + wait_time = between(0.1, 0.3) # Very aggressive + + @task + def rapid_fire(self): + """Send requests as fast as possible.""" + self.make_request(message_type="simple", tag_suffix="-rapid") + + +class SustainedLoadUser(BaseLangflowUser): + """Maintains exactly 1 request/second for steady load testing.""" + + weight = 3 + wait_time = constant_pacing(1) # Exactly 1 request per second per user + + @task + def steady_load(self): + """Send requests at constant 1 RPS per user.""" + self.make_request(message_type="medium", tag_suffix="-steady") + + +class TailLatencyHunter(BaseLangflowUser): + """Mixed workload designed to expose tail latency issues.""" + + weight = 3 + wait_time = between(0.8, 1.5) + + @task + def hunt_tail_latency(self): + """Alternate between simple and complex requests to find tail latency.""" + if random.random() < 0.7: # noqa: S311 + self.make_request(message_type="simple", tag_suffix="-tail") + else: + self.make_request(message_type="large", tag_suffix="-tail-heavy") + + +class ScalabilityTestUser(BaseLangflowUser): + """Tests for scalability limits.""" + + weight = 3 + wait_time = constant(1.0) # Constant load to test scaling + + @task + def scalability_test(self): + """Send medium complexity requests to test scaling limits.""" + self.make_request(message_type="medium", tag_suffix="-scale") + + +class BurstUser(BaseLangflowUser): + """Sends bursts of requests to test connection pooling.""" + + weight = 3 + wait_time = between(5, 10) # Long wait between bursts + + @task + def burst_attack(self): + """Send a burst of 10 requests quickly to test connection handling.""" + for i in range(10): + self.make_request(message_type="minimal", tag_suffix=f"-burst{i}") + gevent.sleep(0.05) # 50ms between requests in burst + + +# Legacy user class for backward compatibility +class FlowRunUser(NormalUser): + """Legacy FlowRunUser - now inherits from NormalUser for backward compatibility.""" + + +# Auto-select shape based on environment variable +_shape_env = os.getenv("SHAPE", "").lower() +_selected = None + +if _shape_env == "stepramp": + _selected = StepRamp +elif _shape_env == "ramp100": + _selected = RampToHundred + +if _selected: + # Create a single exported shape class and remove others so Locust sees only one + SelectedLoadTestShape = type("SelectedLoadTestShape", (_selected,), {}) + globals()["SelectedLoadTestShape"] = SelectedLoadTestShape + + # Remove other shape classes so Locust auto-picks the selected one + for _name, _obj in list(globals().items()): + if ( + inspect.isclass(_obj) + and issubclass(_obj, LoadTestShape) + and _obj is not SelectedLoadTestShape + and _obj is not LoadTestShape + ): + del globals()[_name] diff --git a/src/backend/tests/locust/langflow_run_load_test.py b/src/backend/tests/locust/langflow_run_load_test.py new file mode 100644 index 0000000000..b20a5f70bf --- /dev/null +++ b/src/backend/tests/locust/langflow_run_load_test.py @@ -0,0 +1,353 @@ +#!/usr/bin/env python3 +"""Langflow Load Test Runner + +This script provides an easy way to run Langflow load tests. +For first-time setup, use setup_langflow_test.py to create test credentials. + +Usage: + # First time setup (run once): + python setup_langflow_test.py --interactive + + # Then run load tests: + python run_load_test.py --help + python run_load_test.py --users 10 --duration 60 + python run_load_test.py --shape ramp100 --host http://localhost:7860 +""" + +import argparse +import os +import subprocess +import sys +import time +from pathlib import Path + + +def run_command(cmd, check=True, capture_output=False): + """Run a shell command with proper error handling.""" + print(f"Running: {' '.join(cmd) if isinstance(cmd, list) else cmd}") + try: + if capture_output: + result = subprocess.run(cmd, shell=isinstance(cmd, str), capture_output=True, text=True, check=check) + return result.stdout.strip() + subprocess.run(cmd, shell=isinstance(cmd, str), check=check) + except subprocess.CalledProcessError as e: + print(f"Command failed: {e}") + if capture_output and e.stdout: + print(f"STDOUT: {e.stdout}") + if capture_output and e.stderr: + print(f"STDERR: {e.stderr}") + if check: + sys.exit(1) + + +def check_langflow_running(host): + """Check if Langflow is already running.""" + try: + import httpx + + with httpx.Client(timeout=5.0) as client: + response = client.get(f"{host}/health") + return response.status_code == 200 + except Exception: + return False + + +def test_single_request(host): + """Test a single flow request to ensure the setup works before load testing.""" + import os + + import httpx + + api_key = os.getenv("API_KEY") + flow_id = os.getenv("FLOW_ID") + + if not api_key or not flow_id: + print("โš ๏ธ Missing API_KEY or FLOW_ID for test request") + return False + + print("\n๐Ÿงช Testing single request before load test...") + print(f" Flow ID: {flow_id}") + print(f" API Key: {api_key[:20]}...") + + # First, test basic connectivity + print(" ๐Ÿ”— Testing basic connectivity...") + try: + with httpx.Client(timeout=10.0) as client: + health_response = client.get(f"{host}/health") + if health_response.status_code == 200: + print(" โœ… Health check passed") + else: + print(f" โš ๏ธ Health check failed: {health_response.status_code}") + return False + except Exception as e: + print(f" โŒ Connectivity test failed: {e}") + print(f" Error type: {type(e).__name__}") + return False + + # Now test the actual flow request + print(" ๐ŸŽฏ Testing flow request...") + try: + url = f"{host}/api/v1/run/{flow_id}?stream=false" + payload = { + "input_value": "Hello, this is a test message", + "output_type": "chat", + "input_type": "chat", + "tweaks": {}, + } + headers = {"x-api-key": api_key, "Content-Type": "application/json"} + + with httpx.Client(timeout=30.0) as client: + response = client.post(url, json=payload, headers=headers) + + print(f" ๐Ÿ“ก Response status: {response.status_code}") + + if response.status_code == 200: + try: + data = response.json() + if data.get("outputs"): + print(" โœ… Test request successful - flow is working!") + return True + print(f" โš ๏ธ Flow executed but no outputs returned: {data}") + return False + except Exception as e: + print(f" โš ๏ธ Invalid JSON response: {e}") + return False + else: + print(f" โŒ Test request failed: {response.status_code}") + print(f" Response: {response.text[:200]}...") + return False + + except Exception as e: + print(f" โŒ Test request error: {e}") + return False + + +def wait_for_langflow(host, timeout=60): + """Wait for Langflow to be ready.""" + print(f"Waiting for Langflow to be ready at {host}...") + start_time = time.time() + + while time.time() - start_time < timeout: + if check_langflow_running(host): + print("โœ… Langflow is ready!") + return True + time.sleep(2) + + print(f"โŒ Langflow did not start within {timeout} seconds") + return False + + +def start_langflow(host, port): + """Start Langflow server if not already running.""" + if check_langflow_running(host): + print(f"โœ… Langflow is already running at {host}") + return None + + print(f"Starting Langflow server on port {port}...") + + # Start Langflow in the background + cmd = [ + sys.executable, + "-m", + "langflow", + "run", + "--host", + "0.0.0.0", + "--port", + str(port), + "--auto-login", + "--log-level", + "warning", + ] + + process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + + # Wait for it to be ready + if wait_for_langflow(host, timeout=60): + return process + process.terminate() + return None + + +def run_locust_test(args): + """Run the Locust load test.""" + locust_file = Path(__file__).parent / "langflow_locustfile.py" + + # Check for required environment variables + if not os.getenv("API_KEY"): + print("โŒ API_KEY environment variable not found!") + print("Run langflow_setup_test.py first to create test credentials.") + sys.exit(1) + + if not os.getenv("FLOW_ID"): + print("โŒ FLOW_ID environment variable not found!") + print("Run langflow_setup_test.py first to create test credentials.") + sys.exit(1) + + cmd = [ + "locust", + "-f", + str(locust_file), + "--host", + args.host, + ] + + # Add shape if specified + env = os.environ.copy() + if args.shape: + env["SHAPE"] = args.shape + + # Add other environment variables + env["LANGFLOW_HOST"] = args.host + + if args.headless: + cmd.extend( + [ + "--headless", + "--users", + str(args.users), + "--spawn-rate", + str(args.spawn_rate), + "--run-time", + f"{args.duration}s", + ] + ) + + if args.csv: + cmd.extend(["--csv", args.csv]) + + if args.html: + cmd.extend(["--html", args.html]) + + print(f"\n{'=' * 60}") + print("STARTING LOAD TEST") + print(f"{'=' * 60}") + print(f"Command: {' '.join(cmd)}") + print(f"Host: {args.host}") + print(f"Users: {args.users}") + print(f"Duration: {args.duration}s") + print(f"Shape: {args.shape or 'default'}") + print(f"API Key: {env.get('API_KEY', 'N/A')[:20]}...") + print(f"Flow ID: {env.get('FLOW_ID', 'N/A')}") + if args.html: + print(f"HTML Report: {args.html}") + if args.csv: + print(f"CSV Reports: {args.csv}_*.csv") + print(f"{'=' * 60}\n") + + subprocess.run(cmd, check=False, env=env) + + +def main(): + parser = argparse.ArgumentParser( + description="Run Langflow load tests with automatic setup", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # Run with web UI (interactive) + python run_load_test.py + + # Run headless test with 50 users for 2 minutes + python run_load_test.py --headless --users 50 --duration 120 + + # Run with specific load shape + python run_load_test.py --shape ramp100 --headless --users 100 --duration 180 + + # Run against existing Langflow instance + python run_load_test.py --host http://localhost:8000 --no-start-langflow + + # Save results to CSV + python run_load_test.py --headless --csv results --users 25 --duration 60 + """, + ) + + # Langflow options + parser.add_argument( + "--host", + default="http://localhost:7860", + help="Langflow host URL (default: http://localhost:7860, use https:// for remote instances)", + ) + parser.add_argument("--port", type=int, default=7860, help="Port to start Langflow on (default: 7860)") + parser.add_argument( + "--no-start-langflow", + action="store_true", + help="Don't start Langflow automatically (assume it's already running)", + ) + + # Load test options + parser.add_argument("--headless", action="store_true", help="Run in headless mode (no web UI)") + parser.add_argument("--users", type=int, default=50, help="Number of concurrent users (default: 20)") + parser.add_argument( + "--spawn-rate", type=int, default=2, help="Rate to spawn users at (users per second, default: 2)" + ) + parser.add_argument("--duration", type=int, default=60, help="Test duration in seconds (default: 60)") + parser.add_argument("--shape", choices=["ramp100", "stepramp"], help="Load test shape to use") + parser.add_argument("--csv", help="Save results to CSV files with this prefix") + parser.add_argument("--html", help="Generate HTML report with this filename (e.g., report.html)") + + args = parser.parse_args() + + # Check dependencies + try: + import httpx + import locust + except ImportError as e: + print(f"โŒ Missing dependency: {e}") + print("Install with: pip install locust httpx") + sys.exit(1) + + langflow_process = None + + try: + # Start Langflow if needed + if not args.no_start_langflow: + if args.host.startswith("https://") or not args.host.startswith("http://localhost"): + print(f"โš ๏ธ Remote host detected: {args.host}") + print(" For remote instances, use --no-start-langflow flag") + print(" Example: --host https://your-remote-instance.com --no-start-langflow") + sys.exit(1) + + langflow_process = start_langflow(args.host, args.port) + if not langflow_process: + print("โŒ Failed to start Langflow") + sys.exit(1) + # Just check if it's running + elif not check_langflow_running(args.host): + print(f"โŒ Langflow is not running at {args.host}") + if args.host.startswith("https://"): + print(" Make sure your remote Langflow instance is accessible") + else: + print("Either start Langflow manually or remove --no-start-langflow flag") + sys.exit(1) + else: + print(f"๐Ÿ”— Using existing Langflow instance at {args.host}") + if args.host.startswith("https://"): + print(" โœ… Remote instance mode") + + # Test a single request before running the full load test + if not test_single_request(args.host): + print("โŒ Single request test failed. Aborting load test.") + sys.exit(1) + + # Run the load test + run_locust_test(args) + + except KeyboardInterrupt: + print("\nโš ๏ธ Test interrupted by user") + except Exception as e: + print(f"โŒ Error: {e}") + sys.exit(1) + finally: + # Clean up Langflow process + if langflow_process: + print("\nStopping Langflow server...") + langflow_process.terminate() + try: + langflow_process.wait(timeout=10) + except subprocess.TimeoutExpired: + langflow_process.kill() + print("โœ… Langflow server stopped") + + +if __name__ == "__main__": + main() diff --git a/src/backend/tests/locust/langflow_setup_test.py b/src/backend/tests/locust/langflow_setup_test.py new file mode 100644 index 0000000000..5872850d72 --- /dev/null +++ b/src/backend/tests/locust/langflow_setup_test.py @@ -0,0 +1,556 @@ +#!/usr/bin/env python3 +"""Langflow Load Test Setup CLI + +This script sets up a complete Langflow test environment by: +1. Starting Langflow (optional) +2. Creating a test user account +3. Authenticating and getting JWT tokens +4. Creating API keys +5. Loading a real starter project flow +6. Providing credentials for load testing + +Usage: + python setup_langflow_test.py --help + python setup_langflow_test.py --interactive + python setup_langflow_test.py --flow "Basic Prompting" + python setup_langflow_test.py --list-flows +""" + +import argparse +import asyncio +import json +import sys +import time + + +async def get_starter_projects_from_api(host: str, access_token: str) -> list[dict]: + """Get starter projects from Langflow API.""" + import httpx + + # Ensure proper URL formatting + base_host = host.rstrip("/") + url = f"{base_host}/api/v1/starter-projects/" + print(f" ๐Ÿ” Fetching starter projects from: {url}") + + try: + async with httpx.AsyncClient() as client: + # Try with authentication first + response = await client.get( + url, + headers={"Authorization": f"Bearer {access_token}"}, + timeout=30.0, + ) + + print(f" ๐Ÿ“ก Response status: {response.status_code}") + + # If auth fails, try without authentication (some endpoints might be public) + if response.status_code == 401: + print(" ๐Ÿ”„ Trying without authentication...") + response = await client.get(url, timeout=30.0) + print(f" ๐Ÿ“ก Response status (no auth): {response.status_code}") + + if response.status_code != 200: + print(f"โš ๏ธ Failed to get starter projects: {response.status_code}") + print(f"Response: {response.text}") + return [] + + # Check if response is empty + if not response.text.strip(): + print("โš ๏ธ Empty response from starter projects endpoint") + return [] + + data = response.json() + print(f" โœ… Found {len(data)} starter projects") + return data + + except Exception as e: + print(f"โš ๏ธ Error fetching starter projects: {e}") + if hasattr(e, "response"): + print(f" Status code: {e.response.status_code}") + print(f" Response text: {e.response.text}") + return [] + + +async def list_available_flows(host: str, access_token: str) -> list[tuple[str, str, str]]: + """List all available starter project flows from Langflow API. + + Args: + host: Langflow host URL + access_token: JWT access token for authentication + + Returns: + List of tuples: (flow_name, flow_name, description) + """ + projects = await get_starter_projects_from_api(host, access_token) + + # Known starter project names and descriptions based on the source code + known_projects = [ + ( + "Basic Prompting", + "Basic Prompting", + "A simple chat interface with OpenAI that answers like a pirate โœ… Great for load testing", + ), + ( + "Blog Writer", + "Blog Writer", + "Generate blog posts using web content as reference material โœ… Good for load testing", + ), + ( + "Document Q&A", + "Document Q&A", + "Question and answer system for document content โš ๏ธ Requires file uploads - not ideal for load testing", + ), + ("Memory Chatbot", "Memory Chatbot", "Chatbot with conversation memory using context โœ… Good for load testing"), + ( + "Vector Store RAG", + "Vector Store RAG", + "Retrieval-Augmented Generation with vector storage โš ๏ธ May require setup - test first", + ), + ] + + # Return the known projects if we have the expected number + if len(projects) == len(known_projects): + return known_projects + + # Fallback: generate names based on project count + flows: list[tuple[str, str, str]] = [] + for i, project in enumerate(projects): + if i < len(known_projects): + flow_name, name, description = known_projects[i] + else: + flow_name = f"Starter Project {i + 1}" + name = flow_name + description = "Starter project flow" + + flows.append((flow_name, name, description)) + + return flows + + +async def get_flow_data_by_name(host: str, access_token: str, flow_name: str) -> dict | None: + """Get flow data for a specific starter project by name. + + Args: + host: Langflow host URL + access_token: JWT access token for authentication + flow_name: Name of the flow to retrieve + + Returns: + Flow data as dictionary, or None if not found + """ + projects = await get_starter_projects_from_api(host, access_token) + flows = await list_available_flows(host, access_token) + + # Find the project by name and get its index + for i, (fname, name, _) in enumerate(flows): + if name == flow_name: + if i < len(projects): + # Add the name and description to the project data + project_data = projects[i].copy() + project_data["name"] = name + project_data["description"] = flows[i][2] + return project_data + + print(f"โš ๏ธ Flow '{flow_name}' not found in starter projects") + return None + + +async def select_flow_interactive(host: str, access_token: str) -> str | None: + """Interactive flow selection.""" + flows = await list_available_flows(host, access_token) + + if not flows: + print("โŒ No starter project flows found!") + return None + + print(f"\n{'=' * 80}") + print("AVAILABLE STARTER PROJECT FLOWS") + print(f"{'=' * 80}") + + for i, (flow_name, name, description) in enumerate(flows, 1): + print(f"{i:2d}. {name}") + print(f" {description[:70]}{'...' if len(description) > 70 else ''}") + print() + + while True: + try: + choice = input(f"Select a flow (1-{len(flows)}) or 'q' to quit: ").strip() + if choice.lower() == "q": + return None + + choice_num = int(choice) + if 1 <= choice_num <= len(flows): + selected_flow = flows[choice_num - 1] + print(f"\nโœ… Selected: {selected_flow[1]}") + return selected_flow[0] # Return flow name + print(f"Please enter a number between 1 and {len(flows)}") + except ValueError: + print("Please enter a valid number or 'q' to quit") + except KeyboardInterrupt: + print("\n\nSetup cancelled by user") + return None + + +async def setup_langflow_environment(host: str, flow_name: str | None = None, interactive: bool = False) -> dict: + """Set up complete Langflow environment with real starter project flows.""" + try: + import httpx + except ImportError: + print("โŒ Missing dependency: httpx") + print("Install with: pip install httpx") + sys.exit(1) + + # Configuration - use default Langflow credentials + username = "langflow" + password = "langflow" + + setup_state = { + "host": host, + "username": username, + "password": password, + "user_id": None, + "access_token": None, + "api_key": None, + "flow_id": None, + "flow_name": None, + "flow_data": None, + } + + async with httpx.AsyncClient(base_url=host, timeout=60.0) as client: + # Step 1: Health check + print(f"\n1. Checking Langflow health at {host}...") + try: + health_response = await client.get("/health") + if health_response.status_code != 200: + raise Exception(f"Health check failed: {health_response.status_code}") + print(" โœ… Langflow is running and accessible") + except Exception as e: + print(f" โŒ Health check failed: {e}") + raise + + # Step 2: Skip user creation, use default credentials + print("2. Using default Langflow credentials...") + print(f" โœ… Using username: {username}") + + # Step 3: Login to get JWT token + print("3. Authenticating...") + login_data = { + "username": username, + "password": password, + } + + try: + login_response = await client.post( + "/api/v1/login", + data=login_data, # OAuth2PasswordRequestForm expects form data + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + if login_response.status_code != 200: + raise Exception(f"Login failed: {login_response.status_code} - {login_response.text}") + + tokens = login_response.json() + setup_state["access_token"] = tokens["access_token"] + print(" โœ… Authentication successful") + except Exception as e: + print(f" โŒ Authentication failed: {e}") + raise + + # Step 4: Create API key + print("4. Creating API key...") + headers = {"Authorization": f"Bearer {setup_state['access_token']}"} + + try: + api_key_data = {"name": f"Load Test Key - {int(time.time())}"} + api_key_response = await client.post("/api/v1/api_key/", json=api_key_data, headers=headers) + + if api_key_response.status_code != 200: + raise Exception(f"API key creation failed: {api_key_response.status_code} - {api_key_response.text}") + + api_key_info = api_key_response.json() + setup_state["api_key"] = api_key_info["api_key"] + print(f" โœ… API key created: {api_key_info['api_key'][:20]}...") + except Exception as e: + print(f" โŒ API key creation failed: {e}") + raise + + # Step 5: Select and load flow from API + print("5. Selecting starter project flow...") + + # Flow selection logic + selected_flow_name = None + if interactive: + selected_flow_name = await select_flow_interactive(host, setup_state["access_token"]) + if not selected_flow_name: + print("No flow selected. Exiting.") + sys.exit(0) + elif flow_name: + # Verify the flow exists in the API + flows = await list_available_flows(host, setup_state["access_token"]) + for fname, name, _ in flows: + if name.lower() == flow_name.lower(): + selected_flow_name = name + break + + if not selected_flow_name: + print(f"โŒ Flow '{flow_name}' not found in starter projects!") + print("Available flows:") + for _, name, _ in flows: + print(f" - {name}") + sys.exit(1) + else: + # Default to Basic Prompting + selected_flow_name = "Basic Prompting" + print(" Using default flow: Basic Prompting") + + # Get flow data from API + flow_data = await get_flow_data_by_name(host, setup_state["access_token"], selected_flow_name) + if not flow_data: + print(f"โŒ Could not load flow data for '{selected_flow_name}'") + sys.exit(1) + + setup_state["flow_name"] = flow_data.get("name", selected_flow_name) + setup_state["flow_data"] = flow_data + + print(f" โœ… Selected flow: {setup_state['flow_name']}") + print(f" Description: {flow_data.get('description', 'No description')}") + + # Step 6: Upload the selected flow + print(f"6. Uploading flow: {setup_state['flow_name']}...") + + try: + # Prepare flow data for upload + # Remove the id to let Langflow generate a new one + flow_upload_data = flow_data.copy() + if "id" in flow_upload_data: + del flow_upload_data["id"] + + # Ensure endpoint_name is unique and valid (only letters, numbers, hyphens, underscores) + import re + + sanitized_name = re.sub(r"[^a-zA-Z0-9_-]", "_", setup_state["flow_name"].lower()) + flow_upload_data["endpoint_name"] = f"loadtest_{int(time.time())}_{sanitized_name}" + + flow_response = await client.post("/api/v1/flows/", json=flow_upload_data, headers=headers) + + if flow_response.status_code != 201: + raise Exception(f"Flow upload failed: {flow_response.status_code} - {flow_response.text}") + + flow_info = flow_response.json() + setup_state["flow_id"] = flow_info["id"] + print(" โœ… Flow uploaded successfully") + print(f" Flow ID: {flow_info['id']}") + print(f" Endpoint: {flow_info.get('endpoint_name', 'N/A')}") + except Exception as e: + print(f" โŒ Flow upload failed: {e}") + raise + + return setup_state + + +def print_setup_results(setup_state: dict): + """Print the setup results in a clear format.""" + print(f"\n{'=' * 80}") + print("SETUP COMPLETE - LOAD TEST CREDENTIALS") + print(f"{'=' * 80}") + print(f"Host: {setup_state['host']}") + print(f"Username: {setup_state['username']}") + print(f"Password: {setup_state['password']}") + print(f"User ID: {setup_state.get('user_id', 'N/A')}") + print(f"JWT Token: {setup_state['access_token'][:50]}..." if setup_state["access_token"] else "N/A") + print(f"API Key: {setup_state['api_key']}") + print(f"Flow ID: {setup_state['flow_id']}") + print(f"Flow Name: {setup_state['flow_name']}") + print(f"{'=' * 80}") + + print("\n๐Ÿ“‹ COPY THESE COMMANDS TO RUN LOAD TESTS:") + print(f"{'=' * 80}") + + # Environment variables for easy copy-paste + print("# Set environment variables:") + print(f"export LANGFLOW_HOST='{setup_state['host']}'") + print(f"export API_KEY='{setup_state['api_key']}'") + print(f"export FLOW_ID='{setup_state['flow_id']}'") + print() + + # Direct locust commands + print("# Run load test with web UI:") + print(f"locust -f locustfile.py --host {setup_state['host']}") + print() + + print("# Run headless load test (50 users, 2 minutes):") + print(f"locust -f locustfile.py --host {setup_state['host']} --headless --users 50 --spawn-rate 5 --run-time 120s") + print() + + print("# Run with load shape:") + print( + f"SHAPE=ramp100 locust -f locustfile.py --host {setup_state['host']} --headless --users 100 --spawn-rate 5 --run-time 180s" + ) + print() + + print("# Or use the runner script:") + print( + f"python run_load_test.py --host {setup_state['host']} --no-start-langflow --headless --users 25 --duration 120" + ) + print() + + print("# Generate HTML report:") + print( + f"python run_load_test.py --host {setup_state['host']} --no-start-langflow --headless --users 50 --duration 180 --html report.html" + ) + + print(f"\n{'=' * 80}") + + +def save_credentials(setup_state: dict, output_file: str): + """Save credentials to a file for later use.""" + credentials = { + "host": setup_state["host"], + "api_key": setup_state["api_key"], + "flow_id": setup_state["flow_id"], + "flow_name": setup_state["flow_name"], + "username": setup_state["username"], + "password": setup_state["password"], + "access_token": setup_state["access_token"], + "created_at": time.time(), + } + + try: + with open(output_file, "w") as f: + json.dump(credentials, f, indent=2) + print(f"\n๐Ÿ’พ Credentials saved to: {output_file}") + except Exception as e: + print(f"โš ๏ธ Could not save credentials: {e}") + + +def main(): + parser = argparse.ArgumentParser( + description="Set up Langflow load test environment with real starter project flows", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # Interactive flow selection + python setup_langflow_test.py --interactive + + # Use specific flow + python setup_langflow_test.py --flow "Memory Chatbot" + + # List available flows + python setup_langflow_test.py --list-flows + + # Setup with custom host + python setup_langflow_test.py --host http://localhost:8000 --interactive + + # Save credentials to file + python setup_langflow_test.py --interactive --save-credentials test_creds.json + """, + ) + + parser.add_argument( + "--host", + default="http://localhost:7860", + help="Langflow host URL (default: http://localhost:7860, use https:// for remote instances)", + ) + parser.add_argument("--flow", help="Name of the starter project flow to use") + parser.add_argument("--interactive", action="store_true", help="Interactive flow selection") + parser.add_argument("--list-flows", action="store_true", help="List available starter project flows and exit") + parser.add_argument("--save-credentials", metavar="FILE", help="Save credentials to a JSON file") + + args = parser.parse_args() + + # List flows and exit + if args.list_flows: + + async def list_flows_only(): + try: + import httpx + except ImportError: + print("โŒ Missing dependency: httpx") + print("Install with: pip install httpx") + sys.exit(1) + + # Quick authentication to access the API + username = "langflow" + password = "langflow" + + async with httpx.AsyncClient(base_url=args.host, timeout=30.0) as client: + # Health check + try: + health_response = await client.get("/health") + if health_response.status_code != 200: + raise Exception(f"Langflow not available at {args.host}") + except Exception as e: + print(f"โŒ Cannot connect to Langflow at {args.host}: {e}") + sys.exit(1) + + # Login to get access token + try: + login_data = {"username": username, "password": password} + login_response = await client.post( + "/api/v1/login", + data=login_data, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + if login_response.status_code != 200: + raise Exception(f"Authentication failed: {login_response.status_code}") + + tokens = login_response.json() + access_token = tokens["access_token"] + + except Exception as e: + print(f"โŒ Authentication failed: {e}") + print("Make sure Langflow is running with default credentials (langflow/langflow)") + sys.exit(1) + + # Get flows from API + flows = await list_available_flows(args.host, access_token) + if not flows: + print("โŒ No starter project flows found!") + sys.exit(1) + + print(f"\n{'=' * 80}") + print("AVAILABLE STARTER PROJECT FLOWS") + print(f"{'=' * 80}") + + for flow_name, name, description in flows: + print(f"๐Ÿ“„ {name}") + print(f" Description: {description}") + print() + + print(f"Total: {len(flows)} flows available") + + asyncio.run(list_flows_only()) + sys.exit(0) + + # Validate arguments + if not args.interactive and not args.flow: + print("โŒ Either --interactive or --flow must be specified") + print("Use --help for more information") + sys.exit(1) + + try: + # Run the setup + setup_state = asyncio.run( + setup_langflow_environment(host=args.host, flow_name=args.flow, interactive=args.interactive) + ) + + # Print results + print_setup_results(setup_state) + + # Save credentials if requested + if args.save_credentials: + save_credentials(setup_state, args.save_credentials) + + print("\n๐Ÿš€ Environment setup complete! You can now run load tests.") + + except KeyboardInterrupt: + print("\n\nโš ๏ธ Setup cancelled by user") + sys.exit(1) + except Exception as e: + print(f"\nโŒ Setup failed: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/src/backend/tests/locust/lfx_serve_locustfile.py b/src/backend/tests/locust/lfx_serve_locustfile.py new file mode 100644 index 0000000000..7d421a8258 --- /dev/null +++ b/src/backend/tests/locust/lfx_serve_locustfile.py @@ -0,0 +1,375 @@ +"""LFX Locust Load Testing File. + +Based on the weakness-focused stress test scripts with additional user behaviors. +Includes production-ready fixes for timing, error handling, and reporting. + +This file tests the LFX API (complex serve), not the Langflow API. + +Usage: + # Run with web UI (recommended) + locust -f locustfile_complex_serve.py --host http://127.0.0.1:8000 + + # Run headless with built-in shape + locust -f locustfile_complex_serve.py --host http://127.0.0.1:8000 --headless --shape RampToHundred + + # Run distributed (master) + locust -f locustfile_complex_serve.py --host http://127.0.0.1:8000 --master + + # Run distributed (worker) + locust -f locustfile_complex_serve.py --host http://127.0.0.1:8000 --worker --master-host=localhost + +Environment Variables: + - FLOW_ID: Flow ID to test (default: 5523731d-5ef3-56de-b4ef-59b0a224fdbc) + - API_KEY: API key for authentication (default: test) + - REQUEST_TIMEOUT: Request timeout in seconds (default: 10) + - SHAPE: Load test shape to use (default: none, options: ramp100) +""" + +import inspect +import json +import os +import random +import time + +import gevent +from locust import FastHttpUser, LoadTestShape, between, constant, constant_pacing, events, task + +# Configuration +FLOW_ID = os.getenv("FLOW_ID", "5523731d-5ef3-56de-b4ef-59b0a224fdbc") +API_KEY = os.getenv("API_KEY", "test") +API_ENDPOINT = f"/flows/{FLOW_ID}/run" + +# Test messages with realistic distribution +TEST_MESSAGES = { + "minimal": "Hi", + "simple": "Can you help me?", + "medium": "I need help understanding how machine learning works in this context.", + "complex": "Please analyze this data: " + "x" * 500 + " and provide detailed insights.", + "large": "Here's a complex scenario: " + "data " * 1000, +} + +# Weighted message distribution for realistic load +MESSAGE_WEIGHTS = [("simple", 50), ("medium", 30), ("minimal", 15), ("complex", 4), ("large", 1)] + + +# Load test shapes +class RampToHundred(LoadTestShape): + """0 -> 100 users at 5 users/sec (20s ramp), then hold until 180s total. + + Matches the TLDR test pattern: 3 minutes, ramping to 100 users. + """ + + spawn_rate = 5 + target_users = 100 + total_duration = 180 # seconds + + def tick(self): + run_time = self.get_run_time() + if run_time >= self.total_duration: + return None + users = min(int(run_time * self.spawn_rate), self.target_users) + return users, self.spawn_rate + + +class StepRamp(LoadTestShape): + """Step ramp for finding performance cliffs. + + Steps every 30 seconds: 5 -> 10 -> 15 -> 20 -> 25 -> 30 -> 35 users. + Each step holds for exactly 30 seconds to measure steady-state performance. + """ + + def tick(self): + run_time = self.get_run_time() + + # Define the step progression with 30-second intervals + steps = [ + (30, 5), # 0-30s: 5 users + (60, 10), # 30-60s: 10 users + (90, 15), # 60-90s: 15 users + (120, 20), # 90-120s: 20 users + (150, 25), # 120-150s: 25 users + (180, 30), # 150-180s: 30 users + (210, 35), # 180-210s: 35 users + (240, 40), # 210-240s: 40 users + (270, 45), # 240-270s: 45 users + (300, 50), # 270-300s: 50 users + ] + + # Find current step + for time_limit, user_count in steps: + if run_time < time_limit: + return user_count, 10 # Fast spawn rate for quick transitions + + return None # End test after 300 seconds + + +# Environment-scoped metrics tracking (fixes the event listener issue) +_env_bags = {} + + +@events.test_start.add_listener +def on_test_start(environment, **_kwargs): + """Initialize per-environment metrics tracking.""" + _env_bags[environment] = { + "slow_10s": 0, + "slow_20s": 0, + } + + +@events.request.add_listener +def on_request(request_type, name, response_time, response_length, exception, context, **kwargs): # noqa: ARG001 + """Track slow requests using Locust's built-in timing.""" + # response_time is in milliseconds from Locust + bag = _env_bags.get(context.get("environment") if context else None) + if bag is None: + # fallback: try the single environment we likely have + if len(_env_bags) == 1: + bag = next(iter(_env_bags.values())) + else: + return + + if exception is None: # Only count successful requests for timing + if response_time > 10_000: # 10 seconds in ms + bag["slow_10s"] += 1 + if response_time > 20_000: # 20 seconds in ms + bag["slow_20s"] += 1 + + +@events.test_stop.add_listener +def on_test_stop(environment, **_kwargs): + """Print comprehensive test summary with performance grading.""" + stats = environment.stats.total + if stats.num_requests == 0: + return + + # Get percentiles and basic stats + stats.get_response_time_percentile(0.50) or 0 + p95 = stats.get_response_time_percentile(0.95) or 0 + stats.get_response_time_percentile(0.99) or 0 + fail_ratio = stats.fail_ratio + getattr(stats, "current_rps", 0.0) + + # Get slow request counts + _env_bags.get(environment, {"slow_10s": 0, "slow_20s": 0}) + + # Performance grading based on production criteria + grade = "A" + issues = [] + + if fail_ratio > 0.01: + grade = "B" + issues.append(f"fail {fail_ratio:.1%}") + if fail_ratio > 0.05: + grade = "C" + if p95 > 10_000: + grade = max(grade, "D") + issues.append(f"p95 {p95 / 1000:.1f}s") + if p95 > 20_000: + grade = "F" + issues.append(f"p95 {p95 / 1000:.1f}s") + + # Production readiness assessment + if grade in ["A", "B"] or grade == "C": + pass + else: + pass + + # Cleanup + _env_bags.pop(environment, None) + + +class BaseLfxUser(FastHttpUser): + """Base class for all LFX API load testing user types.""" + + abstract = True + REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "10")) # Tighter timeout for production + + def on_start(self): + """Called when a user starts before any task is scheduled.""" + self.session_id = f"locust_{self.__class__.__name__}_{id(self)}_{int(time.time())}" + self.request_count = 0 + + def make_request(self, message_type="simple", tag_suffix=""): + """Make a request with proper error handling and timing. + + Uses Locust's built-in response time measurement. + """ + message = TEST_MESSAGES.get(message_type, TEST_MESSAGES["simple"]) + + payload = {"input_value": message, "session_id": f"{self.session_id}_{self.request_count}"} + + headers = {"x-api-key": API_KEY, "Content-Type": "application/json"} + + self.request_count += 1 + name = f"{API_ENDPOINT} [{message_type}{tag_suffix}]" + + with self.client.post( + API_ENDPOINT, + json=payload, + headers=headers, + name=name, + timeout=self.REQUEST_TIMEOUT, + catch_response=True, + ) as response: + # Handle successful responses + if response.status_code == 200: + try: + data = response.json() + except json.JSONDecodeError: + return response.failure("Invalid JSON response") + + # Strictly check for success=True in the response payload + success = data.get("success") + if success is True: + return response.success() + + # Application-level failure - success is False, None, or missing + msg = str(data.get("result", "Unknown error"))[:200] + success_status = f"success={success}" if success is not None else "success=missing" + return response.failure(f"Flow failed ({success_status}): {msg}") + + # Handle specific error cases for better monitoring + if response.status_code in (429, 503): + return response.failure(f"Backpressure/capacity: {response.status_code}") + if response.status_code == 401: + return response.failure("Unauthorized - API key issue") + if response.status_code == 404: + return response.failure("Flow not found - check FLOW_ID") + if response.status_code >= 500: + return response.failure(f"Server error {response.status_code}") + + return response.failure(f"HTTP {response.status_code}") + + +class NormalUser(BaseLfxUser): + """Normal user simulating typical API interactions. + + Based on the main stress test patterns with realistic message distribution. + """ + + weight = 3 + wait_time = between(0.5, 2) # Typical user think time + + @task(80) + def send_message(self): + """Main task: Send a message with weighted distribution.""" + message_type = random.choices([w[0] for w in MESSAGE_WEIGHTS], weights=[w[1] for w in MESSAGE_WEIGHTS], k=1)[0] # noqa: S311 + self.make_request(message_type=message_type) + + @task(15) + def send_burst(self): + """Send a burst of 3 small messages quickly.""" + for i in range(3): + self.make_request(message_type="minimal", tag_suffix=f"-burst{i}") + gevent.sleep(0.1) # Small delay between burst requests + + @task(5) + def send_complex(self): + """Occasionally send complex requests that stress the system.""" + self.make_request(message_type="complex") + + +class AggressiveUser(BaseLfxUser): + """Aggressive user with minimal wait times. + + Tests the system under extreme concurrent load. + """ + + weight = 3 + wait_time = between(0.1, 0.3) # Very aggressive + + @task + def rapid_fire(self): + """Send requests as fast as possible.""" + self.make_request(message_type="simple", tag_suffix="-rapid") + + +class SustainedLoadUser(BaseLfxUser): + """Maintains exactly 1 request/second for steady load testing. + + Based on constant throughput testing patterns. + """ + + weight = 3 + wait_time = constant_pacing(1) # Exactly 1 request per second per user + + @task + def steady_load(self): + """Send requests at constant 1 RPS per user.""" + self.make_request(message_type="medium", tag_suffix="-steady") + + +class TailLatencyHunter(BaseLfxUser): + """Mixed workload designed to expose tail latency issues. + + Alternates between light and heavy requests to stress the system. + """ + + weight = 3 + wait_time = between(0.8, 1.5) + + @task + def hunt_tail_latency(self): + """Alternate between simple and complex requests to find tail latency.""" + if random.random() < 0.7: # noqa: S311 + self.make_request(message_type="simple", tag_suffix="-tail") + else: + self.make_request(message_type="large", tag_suffix="-tail-heavy") + + +class ScalabilityTestUser(BaseLfxUser): + """Tests for the scalability cliff at 30 users. + + Uses patterns that specifically stress concurrency limits. + """ + + weight = 3 + wait_time = constant(1.0) # Constant load to test scaling + + @task + def scalability_test(self): + """Send medium complexity requests to test scaling limits.""" + self.make_request(message_type="medium", tag_suffix="-scale") + + +class BurstUser(BaseLfxUser): + """Sends bursts of 10 requests to test connection pooling. + + Based on connection pool exhaustion test patterns. + """ + + weight = 3 + wait_time = between(5, 10) # Long wait between bursts + + @task + def burst_attack(self): + """Send a burst of 10 requests quickly to test connection handling.""" + for i in range(10): + self.make_request(message_type="minimal", tag_suffix=f"-burst{i}") + gevent.sleep(0.05) # 50ms between requests in burst + + +# Auto-select shape based on environment variable + +_shape_env = os.getenv("SHAPE", "").lower() +_selected = None + +if _shape_env == "stepramp": + _selected = StepRamp +elif _shape_env == "ramp100": + _selected = RampToHundred + +if _selected: + # Create a single exported shape class and remove others so Locust sees only one + class SelectedLoadTestShape(_selected): + pass + + # Remove other shape classes so Locust auto-picks the selected one + for _name, _obj in list(globals().items()): + if ( + inspect.isclass(_obj) + and issubclass(_obj, LoadTestShape) + and _obj is not SelectedLoadTestShape + and _obj is not LoadTestShape + ): + del globals()[_name] diff --git a/src/backend/tests/locust/lfx_step_ramp.py b/src/backend/tests/locust/lfx_step_ramp.py new file mode 100644 index 0000000000..6618427ec6 --- /dev/null +++ b/src/backend/tests/locust/lfx_step_ramp.py @@ -0,0 +1,163 @@ +"""LFX Step Ramp Load Test for Finding Performance Cliffs. + +This file tests the LFX API (complex serve), not the Langflow API. + +Steps every 30 seconds: 5 -> 10 -> 15 -> 20 -> 25 -> 30 -> 35 users. +Each step holds for exactly 30 seconds to measure steady-state performance. +""" + +import json +import os +import time + +from locust import FastHttpUser, LoadTestShape, between, events, task + +# Configuration +FLOW_ID = os.getenv("FLOW_ID", "5523731d-5ef3-56de-b4ef-59b0a224fdbc") +API_KEY = os.getenv("API_KEY", "test") +API_ENDPOINT = f"/flows/{FLOW_ID}/run" + +# Test messages +TEST_MESSAGES = { + "minimal": "Hi", + "simple": "Can you help me?", + "medium": "I need help understanding how machine learning works in this context.", + "complex": "Please analyze this data: " + "x" * 500 + " and provide detailed insights.", + "large": "Here's a complex scenario: " + "data " * 1000, +} + +MESSAGE_WEIGHTS = [("simple", 50), ("medium", 30), ("minimal", 15), ("complex", 4), ("large", 1)] + + +class StepRamp(LoadTestShape): + """Step ramp for finding performance cliffs.""" + + def tick(self): + run_time = self.get_run_time() + + # Step every 30 seconds + steps = [ + (30, 5), # 0-30s: 5 users + (60, 10), # 30-60s: 10 users + (90, 15), # 60-90s: 15 users + (120, 20), # 90-120s: 20 users + (150, 25), # 120-150s: 25 users + (180, 30), # 150-180s: 30 users + (210, 35), # 180-210s: 35 users + (240, 40), # 210-240s: 40 users + (270, 45), # 240-270s: 45 users + (300, 50), # 270-300s: 50 users + ] + + for time_limit, user_count in steps: + if run_time < time_limit: + return user_count, 10 + + return None + + +# Event handlers for metrics +_env_bags = {} + + +@events.test_start.add_listener +def on_test_start(environment, **_kwargs): + _env_bags[environment] = {"slow_10s": 0, "slow_20s": 0} + + +@events.request.add_listener +def on_request(request_type, name, response_time, response_length, exception, context, **kwargs): # noqa: ARG001 + """Track slow requests using Locust's built-in timing.""" + # response_time is in milliseconds from Locust + bag = _env_bags.get(context.get("environment") if context else None) + if bag is None: + # fallback: try the single environment we likely have + if len(_env_bags) == 1: + bag = next(iter(_env_bags.values())) + else: + return + + if exception is None: # Only count successful requests for timing + if response_time > 10_000: # 10 seconds in ms + bag["slow_10s"] += 1 + if response_time > 20_000: # 20 seconds in ms + bag["slow_20s"] += 1 + + +@events.test_stop.add_listener +def on_test_stop(environment, **_kwargs): + stats = environment.stats.total + if stats.num_requests == 0: + return + + p95 = stats.get_response_time_percentile(0.95) or 0 + fail_ratio = stats.fail_ratio + _env_bags.get(environment, {"slow_10s": 0, "slow_20s": 0}) + + if fail_ratio > 0.05 or p95 > 10_000: + pass + else: + pass + + _env_bags.pop(environment, None) + + +class BaseLangflowUser(FastHttpUser): + abstract = True + REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "10")) + + def on_start(self): + self.session_id = f"step_{self.__class__.__name__}_{id(self)}_{int(time.time())}" + self.request_count = 0 + + def make_request(self, message_type="simple", tag_suffix=""): + message = TEST_MESSAGES.get(message_type, TEST_MESSAGES["simple"]) + payload = {"input_value": message, "session_id": f"{self.session_id}_{self.request_count}"} + headers = {"x-api-key": API_KEY, "Content-Type": "application/json"} + self.request_count += 1 + name = f"{API_ENDPOINT} [{message_type}{tag_suffix}]" + + with self.client.post( + API_ENDPOINT, + json=payload, + headers=headers, + name=name, + timeout=self.REQUEST_TIMEOUT, + catch_response=True, + ) as response: + # Handle successful responses + if response.status_code == 200: + try: + data = response.json() + except json.JSONDecodeError: + return response.failure("Invalid JSON response") + + # Strictly check for success=True in the response payload + success = data.get("success") + if success is True: + return response.success() + + # Application-level failure - success is False, None, or missing + msg = str(data.get("result", "Unknown error"))[:200] + success_status = f"success={success}" if success is not None else "success=missing" + return response.failure(f"Flow failed ({success_status}): {msg}") + + if response.status_code in (429, 503): + return response.failure(f"Backpressure: {response.status_code}") + if response.status_code == 401: + return response.failure("Unauthorized") + if response.status_code == 404: + return response.failure("Not Found - possible bad FLOW_ID or misconfiguration") + if response.status_code >= 500: + return response.failure(f"Server error {response.status_code}") + return response.failure(f"HTTP {response.status_code}") + + +class StepTestUser(BaseLangflowUser): + """User class for step ramp testing - sends medium complexity requests.""" + + wait_time = between(1, 2) + + @task + def step_test(self): + self.make_request(message_type="medium", tag_suffix="-step") diff --git a/uv.lock b/uv.lock index 26ce53297e..3dd19fcee8 100644 --- a/uv.lock +++ b/uv.lock @@ -5246,7 +5246,7 @@ dev = [ { name = "httpx", specifier = ">=0.28.1" }, { name = "hypothesis", specifier = ">=6.123.17" }, { name = "ipykernel", specifier = ">=6.29.0" }, - { name = "locust", specifier = ">=2.32.9" }, + { name = "locust", specifier = "~=2.40.5" }, { name = "mypy", specifier = ">=1.11.0" }, { name = "packaging", specifier = ">=24.1,<25.0" }, { name = "pandas-stubs", specifier = ">=2.1.4.231227" }, @@ -5984,7 +5984,7 @@ wheels = [ [[package]] name = "locust" -version = "2.41.5" +version = "2.40.5" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "configargparse" }, @@ -6007,9 +6007,9 @@ dependencies = [ { name = "typing-extensions", marker = "python_full_version < '3.12'" }, { name = "werkzeug" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/96/d1/f6731c8c9af1279542698b93cdea1ed8f1a9f4266914ec5f666fea960481/locust-2.41.5.tar.gz", hash = "sha256:f37338b0016382fd4341fc9b8a8f15a37dbfadaa86512195bef69a8e79c39c24", size = 1415560, upload-time = "2025-10-06T13:22:38.206Z" } +sdist = { url = "https://files.pythonhosted.org/packages/0e/7f/8e8a8a79b5bb6cedb840a41bf2b9e6156a8fb608b143498ee0206d4c8170/locust-2.40.5.tar.gz", hash = "sha256:4332f03ebfac83c115763e462f22f495783a88f1d237ccbd618d5b27863f5053", size = 1412591, upload-time = "2025-09-17T11:17:21.112Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/e1/d0/6219189b5770c210569c88a59ae4e9a1455cfd6989f7dd748a3595d52d62/locust-2.41.5-py3-none-any.whl", hash = "sha256:3ab1311a0921e8a2eafed4c376f7c870724a4e9d89cbba4aac4c30f31a4bc451", size = 1434092, upload-time = "2025-10-06T13:22:35.826Z" }, + { url = "https://files.pythonhosted.org/packages/04/0d/bc2f7bc286bd5737049327d0a3c6086dbe10b2bc3f4fe81115f023cf1240/locust-2.40.5-py3-none-any.whl", hash = "sha256:c44a6c415c5218824895bd652a182a958c27a2ceb8427c835d2f4b90d735579b", size = 1431034, upload-time = "2025-09-17T11:17:18.329Z" }, ] [[package]]