✨ Parameterize file split function in data-process task

backend/apps/data_process_app.py

@@ -58,7 +58,8 @@ async def create_task(request: TaskRequest, authorization: Optional[str] = Heade
         original_filename=request.original_filename,
         authorization=authorization,
         embedding_model_id=request.embedding_model_id,
-        tenant_id=request.tenant_id
+        tenant_id=request.tenant_id,
+        telemetry_context=getattr(request, "telemetry_context", {}) or {},
     )
     return JSONResponse(status_code=HTTPStatus.CREATED, content={"task_id": task_result.id})
 

backend/consts/const.py

@@ -259,18 +259,18 @@ class VectorDatabaseType(str, Enum):
 
 
 # Ray Configuration
+DP_PART_PROCESSOR_COUNT = int(os.getenv("DP_PART_PROCESSOR_COUNT", "3"))
+DP_FILE_SPLIT_SIZE_MB = int(os.getenv("DP_FILE_SPLIT_SIZE_MB", "5"))
 RAY_ACTOR_NUM_CPUS = int(os.getenv("RAY_ACTOR_NUM_CPUS", "2"))
 RAY_DASHBOARD_PORT = int(os.getenv("RAY_DASHBOARD_PORT", "8265"))
 RAY_DASHBOARD_HOST = os.getenv("RAY_DASHBOARD_HOST", "0.0.0.0")
-RAY_NUM_CPUS = int(os.getenv("RAY_NUM_CPUS", "4"))
-RAY_OBJECT_STORE_MEMORY_GB = float(
-    os.getenv("RAY_OBJECT_STORE_MEMORY_GB", "0.25"))
+RAY_NUM_CPUS = DP_PART_PROCESSOR_COUNT * RAY_ACTOR_NUM_CPUS
+RAY_OBJECT_STORE_MEMORY_GB = float(os.getenv("RAY_OBJECT_STORE_MEMORY_GB", "0.25"))
 RAY_TEMP_DIR = os.getenv("RAY_TEMP_DIR", "/tmp/ray")
 RAY_LOG_LEVEL = os.getenv("RAY_LOG_LEVEL", "INFO").upper()
 # Disable plasma preallocation to reduce idle memory usage
 # When set to false, Ray will allocate object store memory on-demand instead of preallocating
-RAY_preallocate_plasma = os.getenv(
-    "RAY_preallocate_plasma", "false").lower() == "true"
+RAY_preallocate_plasma = os.getenv("RAY_preallocate_plasma", "false").lower() == "true"
 
 
 # Service Control Flags
@@ -299,13 +299,11 @@ class VectorDatabaseType(str, Enum):
 QUEUES = os.getenv("QUEUES", "process_q,process_part_q,forward_q")
 # Will be dynamically set based on PID if not provided
 WORKER_NAME = os.getenv("WORKER_NAME")
-WORKER_CONCURRENCY = int(os.getenv("WORKER_CONCURRENCY", "4"))
+WORKER_CONCURRENCY = DP_PART_PROCESSOR_COUNT + 1
 RAY_WARM_ACTOR_POOL_SIZE_PART = int(
     os.getenv("RAY_WARM_ACTOR_POOL_SIZE_PART", "2"))
 RAY_WARM_ACTOR_POOL_SIZE_PROCESS = int(
     os.getenv("RAY_WARM_ACTOR_POOL_SIZE_PROCESS", "1"))
-# Global Ray actor pool (shared by process_q/process_part_q workers)
-RAY_GLOBAL_ACTOR_POOL_SIZE = int(os.getenv("RAY_GLOBAL_ACTOR_POOL_SIZE", "3"))
 RAY_ACTOR_WARM_TIMEOUT_S = float(os.getenv("RAY_ACTOR_WARM_TIMEOUT_S", "60"))
 RAY_GLOBAL_ACTOR_POOL_NAME = os.getenv(
     "RAY_GLOBAL_ACTOR_POOL_NAME", "nexent_global_data_processor_pool")

backend/consts/model.py

@@ -414,6 +414,7 @@ class TaskRequest(BaseModel):
     original_filename: Optional[str] = None
     embedding_model_id: Optional[int] = None
     tenant_id: Optional[str] = None
+    telemetry_context: Dict[str, str] = Field(default_factory=dict)
     additional_params: Dict[str, Any] = Field(default_factory=dict)
 
 

backend/data_process/ray_actors.py

@@ -32,8 +32,25 @@ class DataProcessorRayActor:
     """
 
     def __init__(self):
+        # Ray actors are independent processes and must initialize their own
+        # provider before DataProcessCore creates typed preprocessing spans.
+        try:
+            from utils.monitoring import monitoring_manager
+
+            self._monitoring_manager = monitoring_manager
+            telemetry_enabled = monitoring_manager.is_enabled
+        except Exception:
+            self._monitoring_manager = None
+            telemetry_enabled = False
+            logger.warning(
+                "Knowledge telemetry initialization failed in Ray actor; processing will continue",
+                exc_info=True,
+            )
         logger.info(
-            f"Ray actor initialized using {RAY_ACTOR_NUM_CPUS} CPU cores...")
+            "Ray actor initialized using %s CPU cores; telemetry_enabled=%s",
+            RAY_ACTOR_NUM_CPUS,
+            telemetry_enabled,
+        )
         self._processor = DataProcessCore()
 
     def ping(self) -> bool:
@@ -71,12 +88,22 @@ def _run_file_process(
         process_params: Dict[str, Any],
         log_subject: str,
     ) -> List[Dict[str, Any]]:
-        result = self._processor.file_process(
-            file_data=file_data,
+        from utils.knowledge_telemetry import knowledge_span
+
+        with knowledge_span(
+            "knowledge.process.ray_actor",
+            "process.ray_actor",
+            telemetry_context=process_params.get("telemetry_context"),
             filename=filename,
-            chunking_strategy=chunking_strategy,
-            **process_params
-        )
+            file_size_bytes=len(file_data),
+            task_id=process_params.get("task_id"),
+        ):
+            result = self._processor.file_process(
+                file_data=file_data,
+                filename=filename,
+                chunking_strategy=chunking_strategy,
+                **process_params
+            )
 
         chunks, images_info = self._normalize_processor_result(result)
         if images_info:
@@ -304,15 +331,17 @@ def split_file(
         source: str,
         destination: str,
         task_id: Optional[str] = None,
-        max_size: int = 5 * 1024 * 1024,
+        max_size: Optional[int] = None,
+        target_parts: Optional[int] = None,
         file_data: Optional[bytes] = None,
         **params
     ) -> List[bytes]:
         """
         Split file into parts using DataProcessCore.file_split and return raw bytes list.
         """
         logger.info(
-            f"[RayActor] Splitting file: source='{source}', destination='{destination}', task_id='{task_id}', max_size={max_size}"
+            f"[RayActor] Splitting file: source='{source}', destination='{destination}', "
+            f"task_id='{task_id}', max_size={max_size}, target_parts={target_parts}"
         )
 
         if file_data is None:
@@ -336,6 +365,7 @@ def split_file(
             file_data=file_data,
             filename=source,
             max_size=max_size,
+            target_parts=target_parts,
             **params
         )
         split_elapsed = time.perf_counter() - split_start