diff --git a/.gitea/workflows/test_trigger_modal_triton.yaml b/.gitea/workflows/test_trigger_modal_triton.yaml index a3dc150..1656326 100644 --- a/.gitea/workflows/test_trigger_modal_triton.yaml +++ b/.gitea/workflows/test_trigger_modal_triton.yaml @@ -7,7 +7,7 @@ jobs: deploy-to-modal: runs-on: ubuntu-latest container: - image: gitea-modal-runner:latest + image: gitea-modal-runner:latest # where build the code steps: - name: Checkout code diff --git a/workspace/sprint_1_2/CODEBASE/infra/implementation/triton_run/modal_triton.py b/workspace/sprint_1_2/CODEBASE/infra/implementation/triton_run/modal_triton.py index ff4f09a..199a000 100644 --- a/workspace/sprint_1_2/CODEBASE/infra/implementation/triton_run/modal_triton.py +++ b/workspace/sprint_1_2/CODEBASE/infra/implementation/triton_run/modal_triton.py @@ -25,174 +25,6 @@ triton_image = ( ) app = modal.App("triton-s3-service", image=triton_image) -from fastapi import FastAPI, Response, Request,HTTPException -from fastapi.responses import StreamingResponse # 👈 ADD THIS IMPORT -import httpx -web_app = FastAPI() -# ------------------------------------------------------------- -# FASTAPI PROXY ROUTING (Living inside the container) -# ------------------------------------------------------------- - -@web_app.get("/v2/health/ready") -async def forward_health(): - """Proxies external HTTP REST calls straight to Triton's internal inference engine""" - async with httpx.AsyncClient() as client: - try: - response = await client.get("http://127.0.0.1:8000/v2/health/ready") - return Response(content=response.content, status_code=response.status_code, media_type=response.headers.get("content-type")) - except Exception as e: - return Response(content=f"Triton booting models from S3... Error: {str(e)}", status_code=503) - -@web_app.get("/metrics") -@web_app.get("/") -async def forward_metrics(): - """Proxies external metric calls straight to Triton's internal metrics engine""" - async with httpx.AsyncClient() as client: - try: - response = await client.get("http://127.0.0.1:8002/metrics") - return Response(content=response.content, status_code=response.status_code, media_type=response.headers.get("content-type")) - except Exception as e: - return Response(content=f"Waiting for metrics channel... Error: {str(e)}", status_code=503) - -# 👇 ADD THIS CATCH-ALL ROUTE HERE 👇 -@web_app.api_route("/v2/{path:path}", methods=["GET", "POST"]) -async def proxy_all_triton_request(path: str, request: Request): - import tritonclient.grpc.aio as grpcclient - from tritonclient.grpc import service_pb2, service_pb2_grpc - from tritonclient.grpc import _utils as grpc_utils #InferenceServerClient - import grpc - import numpy as np - # 1. Keep HTTP proxy ONLY for metadata/health checks - if "infer" not in path: - async with httpx.AsyncClient(timeout=60.0) as client: - url = f"http://127.0.0.1:8000/v2/{path}" - headers = dict(request.headers) - headers.pop("host", None) - triton_response = await client.request( - method=request.method, url=url, headers=headers, content=await request.body() - ) - return Response( - content=triton_response.content, - status_code=triton_response.status_code, - headers=dict(triton_response.headers) - ) - - # 2. 🚀 FOR INFERENCE: Convert the incoming HTTP raw body into a gRPC call - if "infer" in path: - try: - # Extract model name from the route path (e.g., v2/models/MODEL_NAME/infer) - parts = path.split("/") - model_name = parts[parts.index("models") + 1] - - # Read incoming raw binary HTTP payload - raw_http_body = await request.body() - - header_length_str = request.headers.get("Inference-Header-Content-Length") - if not header_length_str: - raise HTTPException( - status_code=400, - detail="Missing 'Inference-Header-Content-Length' header required for binary Triton transcoding." - ) - - header_length = int(header_length_str) - - # --- 💥 KSERVE V2 MULTI-PART BODY PARSING --- - # Extract the front JSON metadata and the trailing raw binary tensors - import json - json_bytes = raw_http_body[:header_length] - binary_data = raw_http_body[header_length:] - request_metadata = json.loads(json_bytes.decode('utf-8')) - - # Setup async gRPC connection - triton_url = "127.0.0.1:8001" - - # Configure channels to accept large payload returns (100MB limit override) - max_msg_length = 100 * 1024 * 1024 - channel_options = [ - ('grpc.max_receive_message_length', max_msg_length), - ('grpc.max_send_message_length', max_msg_length), - ] - async with grpc.aio.insecure_channel(triton_url, options=channel_options) as channel: - stub = service_pb2_grpc.GRPCInferenceServiceStub(channel=channel) - - # Construct the native ModelInferRequest protobuf - grpc_request = service_pb2.ModelInferRequest() - grpc_request.model_name = model_name - grpc_request.model_version = "" - - # Populate inputs dynamically from incoming KServe metadata - binary_offset = 0 - for input_tensor in request_metadata.get("inputs", []): - # Correct Protobuf repeated field instantiation via .add() - infer_input = grpc_request.inputs.add() - infer_input.name = input_tensor["name"] - infer_input.datatype = input_tensor["datatype"] - infer_input.shape.extend(input_tensor["shape"]) # Explicit clean integers! - - # Extract the binary slice matching this tensor out of the raw payload block - if "parameters" in input_tensor and "binary_data_size" in input_tensor["parameters"]: - data_size = input_tensor["parameters"]["binary_data_size"] - grpc_request.raw_input_contents.append( - binary_data[binary_offset : binary_offset + data_size] - ) - binary_offset += data_size - - # Request output tensor mappings dynamically based on what the client requested - for output_tensor in request_metadata.get("outputs", []): - infer_output = grpc_request.outputs.add() - infer_output.name = output_tensor["name"] - # Signal Triton to return output via raw binary buffers - infer_output.parameters["binary_data"].bool_param = True - - # ✅ Send the transcoding payload straight into Triton over internal gRPC loop - grpc_response = await stub.ModelInfer(request=grpc_request, timeout=None) - - # --- 💥 TRANSCODE gRPC RESPONSE BACK TO MULTI-PART KSERVE HTTP --- - response_metadata = { - "model_name": grpc_response.model_name, - "model_version": grpc_response.model_version, - "outputs": [] - } - - response_binary_body = b"" - for i, output in enumerate(grpc_response.outputs): - out_desc = { - "name": output.name, - "datatype": output.datatype, - "shape": list(output.shape), - "parameters": { - "binary_data_size": len(grpc_response.raw_output_contents[i]) - } - } - response_metadata["outputs"].append(out_desc) - response_binary_body += grpc_response.raw_output_contents[i] - - # Re-bundle into [JSON metadata] + [Raw binary output chunks] - response_json_bytes = json.dumps(response_metadata).encode('utf-8') - output_http_body = response_json_bytes + response_binary_body - - return Response( - content=output_http_body, - status_code=200, - headers={ - "Content-Type": "application/octet-stream", - "Inference-Header-Content-Length": str(len(response_json_bytes)) - } - ) - - except Exception as e: - import traceback - print(f"CRITICAL TRANSLATION EXCEPTION: {traceback.format_exc()}") - return Response( - content=f"Internal gRPC Pipeline Multiplex Error: {str(e)}", - status_code=502 - ) - -@web_app.get("/v2/models") -async def forward_list_models(): - async with httpx.AsyncClient() as client: - r = await client.get("http://127.0.0.1:8000/v2/models") - return Response(content=r.content, status_code=r.status_code, media_type=r.headers.get("content-type")) # ------------------------------------------------------------- # THE UNIFIED SERVICE FUNCTION (1 Container, 1 GPU, 1 Triton Process) @@ -212,6 +44,176 @@ async def forward_list_models(): ) @modal.asgi_app() def unified_triton_server(): + + from fastapi import FastAPI, Response, Request,HTTPException + from fastapi.responses import StreamingResponse # 👈 ADD THIS IMPORT + import httpx + web_app = FastAPI() + # ------------------------------------------------------------- + # FASTAPI PROXY ROUTING (Living inside the container) + # ------------------------------------------------------------- + + @web_app.get("/v2/health/ready") + async def forward_health(): + """Proxies external HTTP REST calls straight to Triton's internal inference engine""" + async with httpx.AsyncClient() as client: + try: + response = await client.get("http://127.0.0.1:8000/v2/health/ready") + return Response(content=response.content, status_code=response.status_code, media_type=response.headers.get("content-type")) + except Exception as e: + return Response(content=f"Triton booting models from S3... Error: {str(e)}", status_code=503) + + @web_app.get("/metrics") + @web_app.get("/") + async def forward_metrics(): + """Proxies external metric calls straight to Triton's internal metrics engine""" + async with httpx.AsyncClient() as client: + try: + response = await client.get("http://127.0.0.1:8002/metrics") + return Response(content=response.content, status_code=response.status_code, media_type=response.headers.get("content-type")) + except Exception as e: + return Response(content=f"Waiting for metrics channel... Error: {str(e)}", status_code=503) + + # 👇 ADD THIS CATCH-ALL ROUTE HERE 👇 + @web_app.api_route("/v2/{path:path}", methods=["GET", "POST"]) + async def proxy_all_triton_request(path: str, request: Request): + import tritonclient.grpc.aio as grpcclient + from tritonclient.grpc import service_pb2, service_pb2_grpc + from tritonclient.grpc import _utils as grpc_utils #InferenceServerClient + import grpc + import numpy as np + # 1. Keep HTTP proxy ONLY for metadata/health checks + if "infer" not in path: + async with httpx.AsyncClient(timeout=60.0) as client: + url = f"http://127.0.0.1:8000/v2/{path}" + headers = dict(request.headers) + headers.pop("host", None) + triton_response = await client.request( + method=request.method, url=url, headers=headers, content=await request.body() + ) + return Response( + content=triton_response.content, + status_code=triton_response.status_code, + headers=dict(triton_response.headers) + ) + + # 2. 🚀 FOR INFERENCE: Convert the incoming HTTP raw body into a gRPC call + if "infer" in path: + try: + # Extract model name from the route path (e.g., v2/models/MODEL_NAME/infer) + parts = path.split("/") + model_name = parts[parts.index("models") + 1] + + # Read incoming raw binary HTTP payload + raw_http_body = await request.body() + + header_length_str = request.headers.get("Inference-Header-Content-Length") + if not header_length_str: + raise HTTPException( + status_code=400, + detail="Missing 'Inference-Header-Content-Length' header required for binary Triton transcoding." + ) + + header_length = int(header_length_str) + + # --- 💥 KSERVE V2 MULTI-PART BODY PARSING --- + # Extract the front JSON metadata and the trailing raw binary tensors + import json + json_bytes = raw_http_body[:header_length] + binary_data = raw_http_body[header_length:] + request_metadata = json.loads(json_bytes.decode('utf-8')) + + # Setup async gRPC connection + triton_url = "127.0.0.1:8001" + + # Configure channels to accept large payload returns (100MB limit override) + max_msg_length = 100 * 1024 * 1024 + channel_options = [ + ('grpc.max_receive_message_length', max_msg_length), + ('grpc.max_send_message_length', max_msg_length), + ] + async with grpc.aio.insecure_channel(triton_url, options=channel_options) as channel: + stub = service_pb2_grpc.GRPCInferenceServiceStub(channel=channel) + + # Construct the native ModelInferRequest protobuf + grpc_request = service_pb2.ModelInferRequest() + grpc_request.model_name = model_name + grpc_request.model_version = "" + + # Populate inputs dynamically from incoming KServe metadata + binary_offset = 0 + for input_tensor in request_metadata.get("inputs", []): + # Correct Protobuf repeated field instantiation via .add() + infer_input = grpc_request.inputs.add() + infer_input.name = input_tensor["name"] + infer_input.datatype = input_tensor["datatype"] + infer_input.shape.extend(input_tensor["shape"]) # Explicit clean integers! + + # Extract the binary slice matching this tensor out of the raw payload block + if "parameters" in input_tensor and "binary_data_size" in input_tensor["parameters"]: + data_size = input_tensor["parameters"]["binary_data_size"] + grpc_request.raw_input_contents.append( + binary_data[binary_offset : binary_offset + data_size] + ) + binary_offset += data_size + + # Request output tensor mappings dynamically based on what the client requested + for output_tensor in request_metadata.get("outputs", []): + infer_output = grpc_request.outputs.add() + infer_output.name = output_tensor["name"] + # Signal Triton to return output via raw binary buffers + infer_output.parameters["binary_data"].bool_param = True + + # ✅ Send the transcoding payload straight into Triton over internal gRPC loop + grpc_response = await stub.ModelInfer(request=grpc_request, timeout=None) + + # --- 💥 TRANSCODE gRPC RESPONSE BACK TO MULTI-PART KSERVE HTTP --- + response_metadata = { + "model_name": grpc_response.model_name, + "model_version": grpc_response.model_version, + "outputs": [] + } + + response_binary_body = b"" + for i, output in enumerate(grpc_response.outputs): + out_desc = { + "name": output.name, + "datatype": output.datatype, + "shape": list(output.shape), + "parameters": { + "binary_data_size": len(grpc_response.raw_output_contents[i]) + } + } + response_metadata["outputs"].append(out_desc) + response_binary_body += grpc_response.raw_output_contents[i] + + # Re-bundle into [JSON metadata] + [Raw binary output chunks] + response_json_bytes = json.dumps(response_metadata).encode('utf-8') + output_http_body = response_json_bytes + response_binary_body + + return Response( + content=output_http_body, + status_code=200, + headers={ + "Content-Type": "application/octet-stream", + "Inference-Header-Content-Length": str(len(response_json_bytes)) + } + ) + + except Exception as e: + import traceback + print(f"CRITICAL TRANSLATION EXCEPTION: {traceback.format_exc()}") + return Response( + content=f"Internal gRPC Pipeline Multiplex Error: {str(e)}", + status_code=502 + ) + + @web_app.get("/v2/models") + async def forward_list_models(): + async with httpx.AsyncClient() as client: + r = await client.get("http://127.0.0.1:8000/v2/models") + return Response(content=r.content, status_code=r.status_code, media_type=r.headers.get("content-type")) + print("🚀 Booting ONE Triton Instance inside ONE A100 Container...") # Spawns Triton in the background. It will automatically read