msk_admin/Lumina-MSK
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update the modal deployment logic
All checks were successful
Triton Modal Trigger / deploy-to-modal (push) Successful in 11s
Details

Browse Source
This commit is contained in:
DatTT127
2026-07-17 13:51:58 +07:00
parent 196d243e03
commit 7d5c583475
2 changed files with 171 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

338
workspace/sprint_1_2/CODEBASE/infra/implementation/triton_run/modal_triton.py
View File

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