pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-27 10:34:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add new Locust load testing with user behavior simulations and error handling improvements (#9848)

Browse Source 
* feat: add enhanced Langflow Locust load testing file with user behavior simulations

* fix: resolve ruff linting errors in Locust test files

- Remove shebang lines from locust files to fix EXE001 errors
- Fix docstring formatting (D205) by adding blank lines after summary
- Remove unused function arguments (ARG001) by prefixing with underscore
- Add noqa comments for acceptable use of random module (S311)
- Move imports to top of file to fix E402

* feat: add enhanced load testing targets with improved error handling and reporting options

* fix: update locustfile references to use the correct filename for load testing

* fix: add error handling for 404 responses in BaseLangflowUser requests

* fix: add missing color definition for yellow in Makefile

* fix: improve request handling in locustfile for better error reporting

- Refactor on_request function parameters for clarity.
- Implement strict checking for success in response payload.
- Enhance failure messages to include success status for better debugging.

* fix: enhance request tracking and error reporting in locustfile

- Refactor on_request function to improve environment handling.
- Implement strict checks for success in response payloads.
- Update failure messages to include success status for better clarity.

* fix: remove unnecessary shape parameter from ramp load test command in Makefile

- Simplified the command for generating HTML reports and running headless tests by removing the shape parameter, ensuring cleaner execution.

* refactor: Clean up import statements across various test files

- Removed unnecessary blank lines and organized import statements for consistency.
- Ensured that all imports are correctly formatted and follow the project's style guidelines.
- Improved readability and maintainability of the test files by standardizing import practices.

* chore: Update Ruff configuration for locust tests

- Added specific linting rules to the Ruff configuration for locust test files to accommodate common CLI practices.
- Included exceptions for missing docstrings, print statements, subprocess calls, and other relevant patterns to enhance code quality while maintaining usability in CLI contexts.
- Improved overall maintainability and clarity of the testing codebase by aligning linting rules with project requirements.

* refactor: Clean up and organize import statements in test files

- Removed unnecessary blank lines and organized imports for consistency across various test files.
- Improved readability and maintainability by standardizing import practices.
- Ensured that all imports are correctly formatted and follow the project's style guidelines.

* feat: Implement comprehensive load testing for Langflow API

- Added a detailed locustfile for Langflow API load testing, including multiple user behaviors and performance analysis.
- Introduced realistic message distributions and enhanced error logging mechanisms.
- Implemented various load test shapes (RampToHundred, StepRamp) to simulate different user loads.
- Improved error handling and logging for better diagnostics during tests.
- Captured Langflow server logs and provided a summary of errors encountered during testing.
- Ensured environment variable configurations for flexibility in testing scenarios.

* refactor: rename lfx locust files

* feat: Introduce a new Langflow API load testing framework

- Added a comprehensive locustfile for Langflow API load testing, replacing the previous locustfile.
- Implemented multiple user behaviors and performance analysis with realistic message distributions.
- Enhanced error logging and metrics tracking for improved diagnostics during tests.
- Introduced various load test shapes (RampToHundred, StepRamp) to simulate different user loads.
- Captured Langflow server logs and provided detailed error summaries for better analysis.
- Configured environment variables for flexible testing scenarios.

* feat: Add Langflow load test setup and runner scripts

- Introduced `langflow_run_load_test.py` for executing load tests with customizable parameters and automatic Langflow server management.
- Added `langflow_setup_test.py` to facilitate the setup of a Langflow test environment, including user account creation and API key generation.
- Implemented interactive flow selection and error handling for a smoother user experience.
- Enhanced documentation with usage examples and detailed descriptions for both scripts.

* feat: Add example Langflow load testing workflow script

- Introduced `langflow_example_workflow.py` to demonstrate a complete workflow for setting up and executing Langflow load tests.
- Implemented dependency checks, user prompts, and environment variable management for a seamless testing experience.
- Enhanced documentation within the script to guide users through the load testing process and available options.

* feat: Add comprehensive README for Langflow load testing

- Introduced a detailed README.md for the Langflow load testing framework using Locust.
- Documented setup processes, user types, load test shapes, and performance grading.
- Included quick start instructions, advanced usage scenarios, and troubleshooting tips.
- Enhanced clarity and usability for users looking to implement load testing for Langflow and LFX APIs.

* feat: Enhance load testing capabilities in Makefile

- Added new load testing commands for enhanced functionality, including setup, execution, and cleanup for Langflow load tests.
- Introduced commands for quick and stress testing, as well as listing available flows and running complete workflows.
- Updated existing commands to use new locust files for improved performance and reporting.
- Enhanced documentation within the Makefile to guide users through the new load testing features and options.

* feat: Enhance Langflow load test scripts with improved host handling

- Updated argument parsing to clarify the use of the `--host` option, including guidance for remote instances.
- Added checks for remote host detection and provided user feedback on starting Langflow.
- Enhanced health check messages to specify accessibility of the Langflow instance.
- Improved documentation within the scripts to assist users in configuring and running load tests effectively.

* docs: Expand README for remote instance load testing

- Added a new section detailing the setup and execution of load tests against remote Langflow instances.
- Included important notes on best practices for remote testing, such as using HTTPS and monitoring resources.
- Updated examples to reflect remote instance configurations, enhancing clarity for users conducting load tests in varied environments.

* feat: Improve API project fetching with enhanced error handling and response validation

- Updated the `get_starter_projects_from_api` function to ensure proper URL formatting and added detailed logging for request status.
- Implemented fallback logic to attempt fetching projects without authentication if the initial request fails.
- Added checks for empty responses and improved error reporting to include status codes and response texts for better diagnostics.
- Enhanced overall robustness of the API interaction, providing clearer feedback during the fetching process.

* feat: Add remote load testing capabilities to Makefile

- Introduced new commands for setting up and running load tests against remote Langflow instances.
- Added checks for the required LANGFLOW_HOST environment variable and provided user feedback for setup and execution.
- Updated load test help documentation to include instructions for remote testing, enhancing usability for users.
- Improved overall structure and clarity of load testing commands in the Makefile.

* feat: Add single request test for Langflow load testing

- Introduced a new function `test_single_request` to validate the setup by testing a single flow request before executing load tests.
- Implemented connectivity checks and response validation to ensure the Langflow instance is operational.
- Updated the load test script to abort if the single request test fails, enhancing robustness and user feedback.
- Adjusted default user count in load test options from 10 to 50 for improved testing scenarios.

* feat: Update load test scripts to use credentials from JSON files and adjust user counts

* docs: Update README with additional notes for remote testing best practices

* feat: Enhance error logging in load testing scripts

- Updated the `log_detailed_error` function to accept an additional `traceback` parameter for improved error reporting.
- Enhanced exception handling in the `BaseLangflowUser` class to capture detailed error information, including connection issues and timeouts.
- Improved logging of connection errors by providing structured error details in the response text, aiding in diagnostics during load tests.

* docs: Update README to enhance load testing instructions and features

- Added sections for pre-flight testing and enhanced error logging to improve user experience.
- Introduced Makefile commands for easier setup and execution of load tests.
- Updated remote testing instructions and clarified the use of default credentials.
- Improved troubleshooting tips and added a diagnostic tool for remote instances.

* feat: Add diagnostic tool for remote Langflow instances

- Introduced a new script `diagnose_remote.py` to assist in debugging connection issues and performance problems with remote Langflow instances.
- Implemented functions for testing connectivity, flow execution, and running load simulations, providing detailed feedback on each step.
- Added command-line arguments for user flexibility, including options for saving results to a JSON file.
- Enhanced user experience with clear output messages and recommendations based on diagnostic results.

* feat: Enhance load test commands in Makefile for automatic setup and flow selection

- Updated `load_test_run`, `load_test_langflow_quick`, and `load_test_stress` commands to automatically set up credentials if not found.
- Added support for specifying flow names via the `FLOW_NAME` environment variable, improving user experience and flexibility.
- Enhanced error messages to guide users in selecting available flows, ensuring smoother execution of load tests.

* feat: Update flow descriptions and enhance endpoint name sanitization in load testing setup

- Improved descriptions of known starter projects to include load testing suitability and requirements.
- Enhanced the uniqueness and validity of the `endpoint_name` by sanitizing it to allow only letters, numbers, hyphens, and underscores, ensuring better compliance with API standards.

* [autofix.ci] apply automated fixes

* revert: remove unintended formatting changes

* feat: update locust dependency to version 2.40.5 in pyproject.toml and uv.lock

* Update uv.lock

* [autofix.ci] apply automated fixes

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Edwin Jose <edwin.jose@datastax.com>
This commit is contained in:
Gabriel Luiz Freitas Almeida
2025-10-08 17:57:34 -03:00
committed by GitHub
parent 719fc0409d
commit 583db4101a
11 changed files with 3163 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
Makefile
View File

@ -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
######################

30
pyproject.toml
View File

@ -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"]

332
src/backend/tests/locust/README.md Normal file
View File

@ -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 <url> --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
```

258
src/backend/tests/locust/diagnose_remote.py Normal file
View File

@ -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()

284
src/backend/tests/locust/langflow_example_workflow.py Normal file
View File

@ -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()

624
src/backend/tests/locust/langflow_locustfile.py Normal file
View File

@ -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]

353
src/backend/tests/locust/langflow_run_load_test.py Normal file
View File

@ -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()

556
src/backend/tests/locust/langflow_setup_test.py Normal file
View File

@ -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()

375
src/backend/tests/locust/lfx_serve_locustfile.py Normal file
View File

@ -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]

163
src/backend/tests/locust/lfx_step_ramp.py Normal file
View File

@ -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")

8
uv.lock generated
View File

@ -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]]