feat: complete multi-process implementation

Author: LuckyYC

backend/db/db_config.py

@@ -35,6 +35,7 @@ class Config:
 
         env_file = ".env"
         case_sensitive = True
+        extra = "ignore"
 
 
 @lru_cache()

backend/model/task.py

@@ -3,9 +3,10 @@
 Copyright (c) 2025, All Rights Reserved.
 """
 
+import re
 from typing import Dict, List, Optional, Union
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, validator
 from sqlalchemy import Column, DateTime, Float, Integer, String, Text, func
 
 from db.mysql import Base
@@ -83,8 +84,12 @@ class HeaderItem(BaseModel):
         fixed: A boolean indicating if the header is fixed (not used currently).
     """
 
-    key: str
-    value: str
+    key: str = Field(
+        ..., min_length=1, max_length=100, description="Header name (1-100 chars)"
+    )
+    value: str = Field(
+        ..., max_length=1000, description="Header value (max 1000 chars)"
+    )
     fixed: bool = True
 
 
@@ -97,8 +102,12 @@ class CookieItem(BaseModel):
         value: The cookie value.
     """
 
-    key: str
-    value: str
+    key: str = Field(
+        ..., min_length=1, max_length=100, description="Cookie name (1-100 chars)"
+    )
+    value: str = Field(
+        ..., max_length=1000, description="Cookie value (max 1000 chars)"
+    )
 
 
 class CertConfig(BaseModel):
@@ -110,8 +119,12 @@ class CertConfig(BaseModel):
         key_file: Path to the SSL private key file.
     """
 
-    cert_file: Optional[str] = Field(None, description="Path to the certificate file")
-    key_file: Optional[str] = Field(None, description="Path to the private key file")
+    cert_file: Optional[str] = Field(
+        None, max_length=255, description="Path to the certificate file (max 255 chars)"
+    )
+    key_file: Optional[str] = Field(
+        None, max_length=255, description="Path to the private key file (max 255 chars)"
+    )
 
 
 class TaskStopReq(BaseModel):
@@ -130,44 +143,169 @@ class TaskCreateReq(BaseModel):
     Request model for creating a new performance testing task.
     """
 
-    temp_task_id: str
-    name: str = Field(..., description="Name of the task")
-    target_host: str = Field(..., description="Target model API host")
-    api_path: str = Field(default="/chat/completions", description="API path to test")
-    model: Optional[str] = Field(default="", description="Name of the model to test")
+    temp_task_id: str = Field(..., max_length=100, description="Temporary task ID")
+    name: str = Field(..., min_length=1, max_length=100, description="Name of the task")
+    target_host: str = Field(
+        ..., min_length=1, max_length=255, description="Target model API host"
+    )
+    api_path: str = Field(
+        default="/chat/completions", max_length=255, description="API path to test"
+    )
+    model: Optional[str] = Field(
+        default="", max_length=255, description="Name of the model to test"
+    )
     duration: int = Field(
-        default=300, ge=1, description="Duration of the test in seconds"
+        default=300,
+        ge=1,
+        le=172800,
+        description="Duration of the test in seconds (1-48 hours)",
+    )
+    concurrent_users: int = Field(
+        ..., ge=1, le=5000, description="Number of concurrent users (1-5000)"
+    )
+    spawn_rate: int = Field(
+        ge=1, le=100, description="Number of users to spawn per second (1-100)"
     )
-    concurrent_users: int = Field(..., ge=1, description="Number of concurrent users")
-    spawn_rate: int = Field(ge=1, description="Number of users to spawn per second")
     chat_type: Optional[int] = Field(
-        default=0, ge=0, description="Type of chat interaction"
+        default=0,
+        ge=0,
+        le=1,
+        description="Type of chat interaction (0=text, 1=multimodal)",
     )
     stream_mode: bool = Field(
         default=True, description="Whether to use streaming response"
     )
     headers: List[HeaderItem] = Field(
-        default_factory=list, description="List of request headers"
+        default_factory=list,
+        description="List of request headers (max 50)",
     )
     cookies: List[CookieItem] = Field(
-        default_factory=list, description="List of request cookies"
+        default_factory=list,
+        description="List of request cookies (max 50)",
     )
     cert_config: Optional[CertConfig] = Field(
         default=None, description="Certificate configuration"
     )
-    system_prompt: Optional[str] = Field(
-        default="", description="System prompt for the model"
-    )
     request_payload: Optional[str] = Field(
-        default="", description="Custom request payload for non-chat APIs (JSON string)"
+        default="",
+        max_length=50000,
+        description="Custom request payload for non-chat APIs (JSON string, max 50000 chars)",
     )
     field_mapping: Optional[Dict[str, str]] = Field(
         default=None, description="Field mapping configuration for custom APIs"
     )
     test_data: Optional[str] = Field(
-        default="", description="Custom test data in JSONL format or file path"
+        default="",
+        max_length=1000000,
+        description="Custom test data in JSONL format or file path (max 1MB)",
     )
 
+    @validator("name")
+    def validate_name(cls, v):
+        if not v or not v.strip():
+            raise ValueError("Name cannot be empty")
+        if len(v.strip()) > 100:
+            raise ValueError("Name length cannot exceed 100 characters")
+        return v.strip()
+
+    @validator("target_host")
+    def validate_target_host(cls, v):
+        if not v or not v.strip():
+            raise ValueError("API address cannot be empty")
+        v = v.strip()
+        if len(v) > 255:
+            raise ValueError("API address length cannot exceed 255 characters")
+        # 基本URL格式验证
+        if not (v.startswith("http://") or v.startswith("https://")):
+            raise ValueError("API address must start with http:// or https://")
+        return v
+
+    @validator("api_path")
+    def validate_api_path(cls, v):
+        if not v or not v.strip():
+            raise ValueError("API path cannot be empty")
+        v = v.strip()
+        if len(v) > 255:
+            raise ValueError("API path length cannot exceed 255 characters")
+        if not v.startswith("/"):
+            raise ValueError("API path must start with /")
+        return v
+
+    @validator("model")
+    def validate_model(cls, v):
+        if v and len(v.strip()) > 255:
+            raise ValueError("Model name length cannot exceed 255 characters")
+        return v.strip() if v else ""
+
+    @validator("request_payload")
+    def validate_request_payload(cls, v, values):
+        """Ensure request_payload is never empty - auto-generate if needed"""
+        # If request_payload is empty, generate default payload
+        if not v or not v.strip():
+            model = values.get("model", "your-model-name")
+            stream_mode = values.get("stream_mode", True)
+
+            # Generate default payload for chat/completions API
+            default_payload = {
+                "model": model,
+                "stream": stream_mode,
+                "messages": [{"role": "user", "content": "Hi"}],
+            }
+
+            import json
+
+            return json.dumps(default_payload)
+
+        # Validate length
+        if len(v) > 50000:
+            raise ValueError("Request payload length cannot exceed 50000 characters")
+
+        # Validate JSON format
+        try:
+            import json
+
+            json.loads(v.strip())
+        except json.JSONDecodeError:
+            raise ValueError("Request payload must be a valid JSON format")
+
+        return v.strip()
+
+    @validator("headers")
+    def validate_headers(cls, v):
+        if len(v) > 50:
+            raise ValueError("Request header count cannot exceed 50")
+        for header in v:
+            if not header.key or not header.key.strip():
+                raise ValueError("Request header name cannot be empty")
+            if len(header.key.strip()) > 100:
+                raise ValueError(
+                    "Request header name length cannot exceed 100 characters"
+                )
+            if len(header.value) > 1000:
+                raise ValueError(
+                    "Request header value length cannot exceed 1000 characters"
+                )
+        return v
+
+    @validator("cookies")
+    def validate_cookies(cls, v):
+        if len(v) > 50:
+            raise ValueError("Cookie count cannot exceed 50")
+        for cookie in v:
+            if not cookie.key or not cookie.key.strip():
+                raise ValueError("Cookie name cannot be empty")
+            if len(cookie.key.strip()) > 100:
+                raise ValueError("Cookie name length cannot exceed 100 characters")
+            if len(cookie.value) > 1000:
+                raise ValueError("Cookie value length cannot exceed 1000 characters")
+        return v
+
+    @validator("test_data")
+    def validate_test_data(cls, v):
+        if v and len(v) > 1000000:  # 1MB
+            raise ValueError("Test data size cannot exceed 1MB")
+        return v
+
 
 class TaskResultItem(BaseModel):
     """
@@ -320,7 +458,6 @@ class Task(Base):
     status = Column(String(32), nullable=False)
     target_host = Column(String(255), nullable=False)
     model = Column(String(100), nullable=True)
-    system_prompt = Column(Text, nullable=True)
     stream_mode = Column(String(20), nullable=False)
     concurrent_users = Column(Integer, nullable=False)
     spawn_rate = Column(Integer, nullable=False)

backend/service/analysis_service.py

@@ -67,7 +67,7 @@ async def analyze_tasks_svc(
         try:
             ai_config = await get_ai_service_config_internal_svc(request)
         except HTTPException as e:
-            error_msg = "Failed to get AI service configuration. %s" % str(e)
+            error_msg = "Failed to get AI service configuration."
             logger.error(error_msg, exc_info=True)
             return AnalysisResponse(
                 task_ids=task_ids,
@@ -201,9 +201,8 @@ async def analyze_tasks_svc(
 
     except Exception as e:
         # Handle other exceptions - only log error, don't update database
-        error_message = "Analysis failed for tasks %s: %s" % (
-            analysis_request.task_ids,
-            str(e),
+        error_message = (
+            f"Analysis failed for tasks {analysis_request.task_ids}: {str(e)}"
         )
         logger.error(error_message, exc_info=True)
         return AnalysisResponse(
@@ -269,7 +268,7 @@ async def get_analysis_svc(request: Request, task_id: str) -> GetAnalysisRespons
         )
 
     except Exception as e:
-        error_msg = "Failed to retrieve analysis for task %s: %s" % (task_id, str(e))
+        error_msg = f"Failed to retrieve analysis for task {task_id}: {str(e)}"
         logger.error(error_msg, exc_info=True)
         return GetAnalysisResponse(
             data=None,
@@ -323,18 +322,18 @@ async def _call_ai_service(
             model_info_str = json.dumps(model_info, ensure_ascii=False, indent=2)
             prompt = prompt_template.format(model_info=model_info_str)
         except (TypeError, ValueError) as e:
-            error_msg = "Failed to serialize model_info: %s" % str(e)
+            error_msg = f"Failed to serialize model_info: {str(e)}"
             logger.error(error_msg)
             # Try fallback serialization
             try:
                 model_info_str = str(model_info)
                 prompt = prompt_template.format(model_info=model_info_str)
             except Exception as fallback_error:
-                logger.error("Fallback serialization failed: %s" % str(fallback_error))
-                raise Exception("Failed to serialize model_info: %s" % str(e))
+                logger.error(f"Fallback serialization failed: {str(fallback_error)}")
+                raise Exception(f"Failed to serialize model_info: {str(e)}")
         except Exception as format_error:
-            error_msg = "Failed to format prompt: %s" % str(format_error)
-            logger.error("Prompt formatting error: %s" % error_msg)
+            error_msg = f"Failed to format prompt: {str(format_error)}"
+            logger.error(f"Prompt formatting error: {error_msg}")
             raise Exception(error_msg)
     else:
         error_msg = "model_info is required for task analysis"
@@ -373,22 +372,22 @@ async def _call_ai_service(
             raise Exception(error_msg)
 
     except httpx.TimeoutException as e:
-        error_msg = "AI service request timeout: %s" % str(e)
-        logger.error("AI service timeout error: %s" % error_msg)
+        error_msg = f"AI service request timeout: {str(e)}"
+        logger.error(f"AI service timeout error: {error_msg}")
         raise Exception(error_msg)
     except httpx.ConnectError as e:
-        error_msg = "AI service connection error: %s" % str(e)
-        logger.error("AI service connection error: %s" % error_msg)
+        error_msg = f"AI service connection error: {str(e)}"
+        logger.error(f"AI service connection error: {error_msg}")
         raise Exception(error_msg)
     except httpx.HTTPStatusError as e:
-        error_msg = "AI service HTTP error: %s - %s" % (e.response.status_code, str(e))
-        logger.error("AI service HTTP error: %s" % error_msg)
+        error_msg = f"AI service HTTP error: {e.response.status_code} - {str(e)}"
+        logger.error(f"AI service HTTP error: {error_msg}")
         raise Exception(error_msg)
     except httpx.RequestError as e:
-        error_msg = "AI service request failed: %s" % str(e)
-        logger.error("AI service request error: %s" % error_msg)
+        error_msg = f"AI service request failed: {str(e)}"
+        logger.error(f"AI service request error: {error_msg}")
         raise Exception(error_msg)
     except Exception as e:
-        error_msg = "AI service call failed: %s" % str(e)
-        logger.error("AI service general error: %s" % error_msg)
+        error_msg = f"AI service call failed: {str(e)}"
+        logger.error(f"AI service general error: {error_msg}")
         raise Exception(error_msg)

backend/service/log_service.py

@@ -98,7 +98,7 @@ def get_last_n_lines(file_path: str, n: int = 100) -> str:
             return result
 
     except Exception as e:
-        logger.error("Failed to read log file: %s" % str(e))
+        logger.error(f"Failed to read log file: {str(e)}")
         return ""
 
 
@@ -155,7 +155,7 @@ async def get_service_log_svc(service_name: str, offset: int, tail: int):
         file_size = os.path.getsize(log_file_path)
         return LogContentResponse(content=content, file_size=file_size)
     except Exception as e:
-        logger.error("Failed to read log file %s: %s" % (log_file_path, str(e)))
+        logger.error(f"Failed to read log file {log_file_path}: {str(e)}")
         return ErrorResponse.internal_server_error(ErrorMessages.LOG_FILE_READ_FAILED)
 
 
@@ -182,5 +182,5 @@ async def get_task_log_svc(task_id: str, offset: int, tail: int):
         file_size = os.path.getsize(log_file_path)
         return LogContentResponse(content=content, file_size=file_size)
     except Exception as e:
-        logger.error("Failed to read log file %s: %s" % (log_file_path, str(e)))
+        logger.error(f"Failed to read log file {log_file_path}: {str(e)}")
         return ErrorResponse.internal_server_error(ErrorMessages.LOG_FILE_READ_FAILED)