[Fix]Add pd routing error handle (#585)

* [CI] Add prefix aware routing test

Signed-off-by: Rui Zhang <[email protected]>

* [ci] refactor k8s discovery e2e test

Signed-off-by: Rui Zhang <[email protected]>

* [CI] Refactor static discovery testing so that it can support multiple logic

Signed-off-by: Rui Zhang <[email protected]>

* [CI] Add static e2e test for prefixaware

Signed-off-by: Rui Zhang <[email protected]>

* refactor the code

Signed-off-by: Rui Zhang <[email protected]>

* [CI] refactor

Signed-off-by: Rui Zhang <[email protected]>

* [CI] Add multiple routing logic test

Signed-off-by: Rui Zhang <[email protected]>

* [CI] fix bug

Signed-off-by: Rui Zhang <[email protected]>

* hotfix/add error handle in pd routing

Signed-off-by: Rui Zhang <[email protected]>

* modify

Signed-off-by: Rui Zhang <[email protected]>

---------

Signed-off-by: Rui Zhang <[email protected]>

Author: zerofishnoodles

src/vllm_router/services/request_service/request.py

@@ -357,21 +357,75 @@ async def route_disaggregated_prefill_request(
     orig_max_tokens = request_json.get("max_tokens", 0)
     request_json["max_tokens"] = 1
     st = time.time()
-    await send_request_to_prefiller(
-        request.app.state.prefill_client, endpoint, request_json, request_id
-    )
-    et = time.time()
-    logger.info(f"{request_id} prefill time (TTFT): {et - st:.4f}")
-    logger.info(
-        f"Routing request {request_id} with session id None to {request.app.state.prefill_client.base_url} at {et}, process time = {et - in_router_time:.4f}"
-    )
-    request_json["max_tokens"] = orig_max_tokens
+    try:
+        await send_request_to_prefiller(
+            request.app.state.prefill_client, endpoint, request_json, request_id
+        )
+        et = time.time()
+        logger.info(f"{request_id} prefill time (TTFT): {et - st:.4f}")
+        logger.info(
+            f"Routing request {request_id} with session id None to {request.app.state.prefill_client.base_url} at {et}, process time = {et - in_router_time:.4f}"
+        )
+        request_json["max_tokens"] = orig_max_tokens
+    except httpx.HTTPStatusError as e:
+        logger.error(f"HTTP error in prefiller: {e}", exc_info=True)
+        return JSONResponse(
+            status_code=e.response.status_code,
+            content={
+                "error": {
+                    "message": f"Prefiller error: {e.response.text}",
+                    "type": "prefiller_error",
+                    "code": e.response.status_code,
+                }
+            },
+            headers={"X-Request-Id": request_id},
+        )
+    except Exception as e:
+        logger.error(f"Unexpected error in prefiller: {e}", exc_info=True)
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": {
+                    "message": f"Prefiller error: {str(e)}",
+                    "type": "prefiller_error",
+                    "code": 500,
+                }
+            },
+            headers={"X-Request-Id": request_id},
+        )
 
     async def generate_stream():
-        async for chunk in send_request_to_decode(
-            request.app.state.decode_client, endpoint, request_json, request_id
-        ):
-            yield chunk
+        try:
+            async for chunk in send_request_to_decode(
+                request.app.state.decode_client, endpoint, request_json, request_id
+            ):
+                yield chunk
+        except httpx.HTTPStatusError as e:
+            logger.error(f"HTTP error in decoder: {e}", exc_info=True)
+            try:
+                error_text = e.response.text
+            except Exception:
+                error_text = f"HTTP {e.response.status_code}"
+            # Yield error as JSON response
+            error_response = {
+                "error": {
+                    "message": f"Decoder error: {error_text}",
+                    "type": "decoder_error",
+                    "code": e.response.status_code,
+                }
+            }
+            yield json.dumps(error_response).encode("utf-8")
+        except Exception as e:
+            logger.error(f"Unexpected error in decoder: {e}", exc_info=True)
+            # Yield error as JSON response
+            error_response = {
+                "error": {
+                    "message": f"Decoder error: {str(e)}",
+                    "type": "decoder_error",
+                    "code": 500,
+                }
+            }
+            yield json.dumps(error_response).encode("utf-8")
 
     curr_time = time.time()
     logger.info(